Cloud Run 関数を使用するワークフロー

目標:

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

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

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

始める前に

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

プロジェクトを設定する

Google Cloud アカウントにログインします。 Google Cloudを初めて使用する場合は、 アカウントを作成して、実際のシナリオでの Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分を差し上げます。

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

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

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

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, Cloud Storage, and Cloud Run functions APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します。

gcloud init

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

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

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

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, Compute Engine, Cloud Storage, and Cloud Run functions APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Google Cloud CLI をインストールします。

外部 ID プロバイダ（IdP）を使用している場合は、まず連携 ID を使用して gcloud CLI にログインする必要があります。

gcloud CLI を初期化するには、次のコマンドを実行します。

gcloud init

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

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

1. Google Cloud コンソールで、Cloud Storage の [バケット] ページに移動します。

   [バケット] に移動

2. [ 作成] をクリックします。
3. [バケットの作成] ページでユーザーのバケット情報を入力します。次のステップに進むには、[続行] をクリックします。
   1. [始める] セクションで、次の操作を行います。
      • バケット名の要件を満たす、グローバルに一意の名前を入力します。
      • バケットラベルを追加するには、[ラベル] セクション（）を開き、[ ラベルを追加] をクリックして、ラベルの key と value を指定します。
   2. [データの保存場所の選択] セクションで、次の操作を行います。
      1. ロケーション タイプを選択してください。
      2. [ロケーション タイプ] プルダウン メニューから、バケットのデータが永続的に保存されるロケーションを選択します。
         • ロケーション タイプとして [デュアルリージョン] を選択した場合は、関連するチェックボックスを使用してターボ レプリケーションを有効にすることもできます。
      3. クロスバケット レプリケーションを設定するには、[Storage Transfer Service 経由でクロスバケット レプリケーションを追加する] を選択し、次の手順を実施します。

         クロスバケット レプリケーションを設定する

         1. [バケット] メニューで、バケットを選択します。

         2. [レプリケーション設定] セクションで、[構成] をクリックして、レプリケーション ジョブの設定を構成します。

            [クロスバケット レプリケーションを構成する] ペインが表示されます。

            • オブジェクト名の接頭辞で複製するオブジェクトをフィルタするには、オブジェクトを追加または除外する接頭辞を入力し、 [接頭辞を追加] をクリックします。
            • 複製されたオブジェクトのストレージ クラスを設定するには、[ストレージ クラス] メニューからストレージ クラスを選択します。この手順をスキップすると、複製されたオブジェクトはデフォルトで宛先バケットのストレージ クラスを使用します。
            • [完了] をクリックします。
   3. [データの保存場所を選択する] セクションで、次の操作を行います。
      1. バケットのデフォルトのストレージ クラスを選択するか、バケットデータのストレージ クラスを自動的に管理する Autoclass を選択します。
      2. 階層名前空間を有効にするには、[データ量が多いワークロード向けにストレージを最適化] セクションで、[このバケットで階層的な名前空間を有効にする] を選択します。
         注: 既存のバケットで階層型名前空間を有効にすることはできません。
   4. [オブジェクトへのアクセスを制御する方法を選択する] セクションで、バケットに公開アクセスの防止を適用するかどうかを選択し、バケットのオブジェクトに使用するアクセス制御方法を選択します。
      注: この設定が組織のポリシーで適用されている場合、[公開アクセスの防止] 設定を変更することはできません。
   5. [オブジェクト データを保護する方法を選択する] セクションで、次の操作を行います。
      • [データ保護] で、バケットに設定するオプションを選択します。
        • 削除（復元可能）を有効にするには、[削除（復元可能）ポリシー（データ復元用）] チェックボックスをオンにして、削除後にオブジェクトを保持する日数を指定します。
        • オブジェクトのバージョニングを設定するには、[オブジェクトのバージョニング（バージョン管理用）] チェックボックスをオンにして、オブジェクトごとの最大バージョン数と、非現行バージョンの有効期限が切れるまでの日数を指定します。
        • オブジェクトとバケットで保持ポリシーを有効にするには、[保持（コンプライアンス用）] チェックボックスをオンにして、次の操作を行います。
          • オブジェクト保持ロックを有効にするには、[オブジェクト保持を有効にする] チェックボックスをオンにします。
          • バケットロックを有効にするには、[バケット保持ポリシーを設定] チェックボックスをオンにして、保持期間の単位と期間を選択します。
      • オブジェクト データの暗号化方法を選択するには、[データ暗号化] セクション（）を開き、データ暗号化方法を選択します。
4. [作成] をクリックします。

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

ワークフロー テンプレートを作成して定義するには、ローカル ターミナル ウィンドウまたは 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. wordcount ジョブをワークフロー テンプレートに追加します。
   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. ワークフローを実行するには、マネージドの単一ノードクラスタを使用します。Managed Service for Apache Spark によってクラスタが作成され、ワークフローがクラスタ上で実行され、ワークフローの完了時にクラスタが削除されます。

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

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

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

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

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
   

Cloud Functions の関数を作成する

1. Google Cloud コンソールで [Cloud Run 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 コンソールに残ります。Managed Service for Apache Spark の [ジョブ] ページに表示されている count... ジョブをクリックして、ワークフロー ジョブの詳細を表示します。

クリーンアップ

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

プロジェクトを削除する

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

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

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

1. Google Cloud コンソールで [リソースの管理] ページに移動します。

   [リソースの管理] に移動

2. プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
3. ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

Cloud Storage バケットを削除する

1. Google Cloud コンソールで、Cloud Storage の [バケット] ページに移動します。

   [バケット] に移動

2. 削除するバケットのチェックボックスをクリックします。
3. バケットを削除するには、 [削除] をクリックして、指示に沿って操作します。

ワークフロー テンプレートを削除する

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

Cloud Functions の関数を削除する

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

次のステップ

• Managed Service for Apache Spark ワークフロー テンプレートの概要をご覧ください。
• ワークフロー スケジューリング ソリューションを確認する