Cloud Functions を使用するワークフロー

目的:

1. ワードカウント ジョブを実行する Dataproc ワークフロー テンプレートを作成する
2. ファイルが Cloud Storage に追加されたときに、ワードカウント ワークフローをトリガーする Node.js Cloud Functions を作成する

ワークフローは「ファイア アンド フォーゲット」です。ワークフローをトリガーする関数が送信された際、例外がなければ、Dataproc はワークフローを最後まで実行します。トリガーする関数内でワークフローの完了を待つ必要はありません。

注: この関数は Cloud Storage のバケットに新しいファイルを追加することでトリガーされ、ワードカウント バイナリは新しいファイルを作成して Cloud Storage に出力するので、反復（ワードカウントが自身の出力ファイルを処理すること）を避けるため、このチュートリアルでは個別の入力バケットと出力バケットを使用します。

始める前に

まだ設定していない場合は、Google Cloud プロジェクトと 2 つの Cloud Storage バケットを設定します。

プロジェクトを設定する

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. Enable the Dataproc, Compute Engine, Cloud Storage, and Cloud Functions APIs.

   Enable the APIs

5. Install the Google Cloud CLI.

6. To initialize the gcloud CLI, run the following command:

   gcloud init

7. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

8. Make sure that billing is enabled for your Google Cloud project.

9. Enable the Dataproc, Compute Engine, Cloud Storage, and Cloud Functions APIs.

   Enable the APIs

10. Install the Google Cloud CLI.

11. To initialize the gcloud CLI, run the following command:

    gcloud init

プロジェクトで 2 つの Cloud Storage バケットを作成または使用する

プロジェクトには、入力ファイル用と出力ファイル用の 2 つの Cloud Storage バケットが必要です。

1. In the Google Cloud console, go to the Cloud Storage Buckets page.

   Go to Buckets page

2. Click Create bucket.
3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
   • For Name your bucket, enter a name that meets the bucket naming requirements.
   • For Choose where to store your data, do the following:
     • Select a Location type option.
     • Select a Location option.
   • For Choose a default storage class for your data, select a storage class.
   • For Choose how to control access to objects, select an Access control option.
   • For Advanced settings (optional), specify an encryption method, a retention policy, or bucket labels.
4. Click Create.

ワークフロー テンプレートを作成する

ローカル ターミナルウィンドウまたは Cloud Shell で以下に表示されているコマンドをコピーして実行し、ワークフロー テンプレートを作成して定義します。

注:

• このコマンドでは、「us-central1」リージョンを指定しています。以前に gcloud config set compute/region を実行してリージョン プロパティを設定した場合は、別のリージョンを指定するか、--region フラグを削除できます。
• 「-- 」（ダッシュ、ダッシュ、スペース）シーケンスは、jar ファイルに引数を渡します。wordcount input_bucket output_dir コマンドは、Cloud Storage input_bucket に含まれるテキスト ファイルに対して、jar のワードカウント アプリケーションを実行し、ワードカウント ファイルを output_bucket に出力します。関数でこの引数を指定できるように、ワードカウントの入力バケット引数をパラメータ化します。

1. ワークフロー テンプレートを作成します。

   gcloud dataproc workflow-templates create wordcount-template \
       --region=us-central1
   

2. ワードカウント ジョブをワークフロー テンプレートに追加します。
   1. コマンドを実行する前に output-bucket-name を指定します（関数は入力バケットを提供します）。出力バケット名を挿入すると、出力バケット引数は gs://your-output-bucket/wordcount-output" のようになります。
   2. 「count」ステップ ID は必須です。追加された Hadoop ジョブを識別します。

   gcloud dataproc workflow-templates add-job hadoop \
       --workflow-template=wordcount-template \
       --step-id=count \
       --jar=file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar \
       --region=us-central1 \
       -- wordcount gs://input-bucket gs://output-bucket-name/wordcount-output
   

3. ワークフローを実行するには、マネージドの単一ノードクラスタを使用します。Dataproc によってクラスタが作成され、ワークフローがクラスタ上で実行され、ワークフローの完了時にクラスタが削除されます。

   gcloud dataproc workflow-templates set-managed-cluster wordcount-template \
       --cluster-name=wordcount \
       --single-node \
       --region=us-central1
   

4. Google Cloud コンソールの [Dataproc ワークフロー] ページで wordcount-template 名をクリックし、[ワークフロー テンプレートの詳細] ページを開きます。「wordcount-template」属性を確認します。

   

ワークフロー テンプレートをパラメータ化する

ワークフロー テンプレートに渡す入力バケット変数をParameterizeします。

1. パラメータ化のために、ワークフロー テンプレートを wordcount.yaml テキスト ファイルにエクスポートします。

   gcloud dataproc workflow-templates export wordcount-template \
       --destination=wordcount.yaml \
       --region=us-central1
   

2. テキスト エディタを使用して wordcount.yaml を開き、ワークフローがトリガーされた際に、ワードカウント バイナリに args[1] として Cloud Storage の INPUT_BUCKET_URI が渡されるようにするため、YAML ファイルの末尾に parameters ブロックを追加します。

   エクスポートされた YAML ファイルの例を以下に示します。テンプレートを更新する方法は次の 2 つがあります。このうちの 1 つを選びます。

   1. your-output_bucket を出力バケット名に置き換えたあと、エクスポートした wordcount.yaml を置き換えるため、ファイル全体をコピーして貼り付けます。
   2. エクスポートした wordcount.yaml ファイルの末尾に、parameters セクションのみをコピーして貼り付けます。
   .

   jobs:
   - hadoopJob:
       args:
       - wordcount
       - gs://input-bucket
       - gs://your-output-bucket/wordcount-output
       mainJarFileUri: file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar
     stepId: count
   placement:
     managedCluster:
       clusterName: wordcount
       config:
         softwareConfig:
           properties:
             dataproc:dataproc.allow.zero.workers: 'true'
   parameters:
   - name: INPUT_BUCKET_URI
     description: wordcount input bucket URI
     fields:
     - jobs['count'].hadoopJob.args[1]
   

3. パラメータ化された wordcount.yaml テキスト ファイルをインポートします。テンプレートを上書きするように求められたら、「Y」（Yes の Y）を入力します。

   gcloud dataproc workflow-templates import  wordcount-template \
       --source=wordcount.yaml \
       --region=us-central1
   

関数を作成する

1. Google Cloud コンソールで [Cloud Functions] ページを開き、[関数の作成] をクリックします。

2. [関数の作成] ページで、次の情報を入力または選択します。

   1. [名前]: wordcount
   2. [メモリ割り当て]: デフォルトの選択のままにします。
   3. [トリガー]:
      • Cloud Storage
      • [イベント タイプ]: 確定 / 作成
      • [バケット]: 入力バケットを選択します（プロジェクトで Cloud Storage バケットを作成するを参照）。このバケットにファイルが追加されると、この関数はワークフローをトリガーします。ワークフローでワードカウント アプリケーションが実行され、バケット内のすべてのテキスト ファイルが処理されます。

   4. ソースコード:

      • インライン エディタ
      • [ランタイム]: Node.js 8
      • [INDEX.JS タブ]: デフォルトのコード スニペットを、次のコードに置き換えます。そして、-your-project-id-（先頭または末尾に「-」がないもの）を指定するように const projectId 行を編集します。

      const dataproc = require('@google-cloud/dataproc').v1;
      
      exports.startWorkflow = (data) => {
       const projectId = '-your-project-id-'
       const region = 'us-central1'
       const workflowTemplate = 'wordcount-template'
      
      const client = new dataproc.WorkflowTemplateServiceClient({
         apiEndpoint: `${region}-dataproc.googleapis.com`,
      });
      
      const file = data;
      console.log("Event: ", file);
      
      const inputBucketUri = `gs://${file.bucket}/${file.name}`;
      
      const request = {
        name: client.projectRegionWorkflowTemplatePath(projectId, region, workflowTemplate),
        parameters: {"INPUT_BUCKET_URI": inputBucketUri}
      };
      
      client.instantiateWorkflowTemplate(request)
        .then(responses => {
          console.log("Launched Dataproc Workflow:", responses[1]);
        })
        .catch(err => {
          console.error(err);
        });
      };
      

      • [PACKAGE.JSON タブ]: デフォルトのコード スニペットを、次のコードに置き換えます。

      {
        "name": "dataproc-workflow",
        "version": "1.0.0",
        "dependencies":{ "@google-cloud/dataproc": ">=1.0.0"}
      }
      

      • 実行する関数: 「startWorkflow」を挿入します。

   5. [作成] をクリックします。

関数をテストする

1. 公開ファイル rose.txt をバケットにコピーして、関数をトリガーします。コマンドに your-input-bucket-name（関数のトリガーに使用されるバケット）を挿入します。

   gcloud storage cp gs://pub/shakespeare/rose.txt gs://your-input-bucket-name
   

2. 30 秒待ってから、次のコマンドを実行して関数が正常に完了したことを確認します。

   gcloud functions logs read wordcount
   

   ...
   Function execution took 1348 ms, finished with status: 'ok'
   

3. Google Cloud コンソールの [関数リスト] ページから関数ログを表示するには、wordcount 関数名をクリックし、[関数の詳細] ページで [ログを表示] をクリックします。

   

4. 出力バケットの wordcount-output フォルダは、Google Cloud コンソールの [ストレージ ブラウザ] ページから表示できます。

   

   注: wordcount-output フォルダが存在する場合、wordcount ジョブは失敗します。関数を再トリガーしてワークフローを再実行する前に、出力バケットの wordcount-output フォルダを削除します。。

5. ワークフローが完了すると、ジョブの詳細は Google Cloud コンソールに残ります。Dataproc の [ジョブ] ページに表示されているcount...ジョブをクリックして、ワークフロー ジョブの詳細を表示します。

   

クリーンアップ

このチュートリアルのワークフローでは、ワークフローが完了するとマネージド クラスタが削除されます。繰り返し費用が発生しないようにするには、このチュートリアルに関連する他のリソースを削除します。

プロジェクトの削除

注意: プロジェクトを削除すると、次のような影響があります。

• プロジェクト内のすべてのものが削除されます。このドキュメントのタスクで既存のプロジェクトを使用した場合、それを削除すると、そのプロジェクトで行った他の作業もすべて削除されます。
• カスタム プロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタム プロジェクト ID を作成した可能性があります。そのプロジェクト ID を使用した URL（たとえば、appspot.com）を保持するには、プロジェクト全体ではなくプロジェクト内の選択したリソースだけを削除します。

複数のアーキテクチャ、チュートリアル、クイックスタートを実施する予定がある場合は、プロジェクトを再利用すると、プロジェクトの割り当て上限を超えないようにすることができます。

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

Cloud Storage バケットの削除

1. In the Google Cloud console, go to the Cloud Storage Buckets page.

   Go to Buckets

2. Click the checkbox for the bucket that you want to delete.
3. To delete the bucket, click Delete, and then follow the instructions.

ワークフロー テンプレートの削除

gcloud dataproc workflow-templates delete wordcount-template \
    --region=us-central1

Cloud Functions の関数の削除

Google Cloud コンソールで Cloud Functions ページを開き、wordcount 関数の左側にあるボックスを選択して、[削除] をクリックします。

次のステップ

• Dataproc ワークフロー テンプレートの概要を確認する
• ワークフローのスケジューリング ソリューションを確認する