Pub/Sub から BigQuery へのストリーミング

---

このチュートリアルでは、Pub/Sub Subscription to BigQuery テンプレートを使用し、Google Cloud コンソールまたは Google Cloud CLI で Dataflow テンプレート ジョブを作成して実行します。このチュートリアルでは、Pub/Sub から JSON エンコードのメッセージを読み取り、Google 提供のストリーミング テンプレートを拡張するユーザー定義関数（UDF）を使用し、Apache Beam SDK でメッセージ データを変換して、その結果を BigQuery テーブルに書き込むストリーミング パイプラインの例を見ていきます。

ストリーミング分析とデータ インテグレーション パイプラインでは、Pub/Sub を使用してデータの取り込みと配信を行います。Pub/Sub を使用すると、イベント プロデューサー（パブリッシャー）とコンシューマー（サブスクライバー）のシステムを作成できます。パブリッシャーはイベントを Pub/Sub サービスに非同期に送信します。Pub/Sub は、イベントに反応する必要のあるすべてのサービスにイベントを配信します。

Dataflow は、ストリーミング（リアルタイム）モードとバッチモードの両方でデータの変換と拡充を行うためのフルマネージド型のサービスです。Apache Beam SDK を使用して受信データを変換し、変換したデータを出力する簡素化されたパイプライン開発環境を提供します。

このワークフローの利点は、UDF を使用して、BigQuery に書き込まれる前にメッセージ データを変換できることです。別の方法として、BigQuery サブスクリプションを使用します。これにより、Dataflow を使用せずに Pub/Sub メッセージを BigQuery に直接書き込むことができます。このオプションは、at-least-once 配信のみをサポートし、exactly-once 処理はサポートしません。

目標

• Pub/Sub トピックを作成する。
• テーブルとスキーマを使用して BigQuery データセットを作成する。
• Dataflow と Google 提供のストリーミング テンプレートを使用して、Pub/Sub サブスクリプションから BigQuery にデータをストリーミングする。
• Google 提供のストリーミング テンプレートを拡張するユーザー定義関数（UDF）を作成します。

費用

このドキュメントでは、Google Cloud の次の課金対象のコンポーネントを使用します。

• Dataflow
• Pub/Sub
• Cloud Storage
• Cloud Scheduler
• BigQuery

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを生成できます。 新しい Google Cloud ユーザーは無料トライアルをご利用いただける場合があります。

このドキュメントに記載されているタスクの完了後、作成したリソースを削除すると、それ以上の請求は発生しません。詳細については、クリーンアップをご覧ください。

始める前に

このセクションでは、プロジェクトを選択して API を有効にし、ユーザー アカウントとワーカー サービス アカウントに適切なロールを付与する方法について説明します。

コンソール

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 Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs.

   Enable the APIs

5. 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

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

7. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs.

   Enable the APIs

8. このチュートリアルの手順を完了するには、ユーザー アカウントにサービス アカウント ユーザーのロールが必要です。Compute Engine のデフォルトのサービス アカウントには、Dataflow ワーカー、Dataflow 管理者、Pub/Sub 編集者、Storage オブジェクト管理者、BigQuery データ編集者のロールが付与されている必要があります。Google Cloud コンソールで必要なロールを追加する手順は次のとおりです。

   1. Google Cloud コンソールの [IAM] ページに移動します。

      IAM に移動
   2. プロジェクトを選択します。
   3. ユーザー アカウントを含む行で、「プリンシパルを編集します」アイコン をクリックし、[ 別のロールを追加] をクリックします。
   4. プルダウン リストで、[サービス アカウント ユーザー] のロールを選択します。
   5. Compute Engine のデフォルトのサービス アカウントを含む行で、「プリンシパルを編集します」アイコン をクリックし、[ 別のロールを追加] をクリックします。
   6. プルダウン リストで、[Dataflow ワーカー] ロールを選択します。

   7. この手順を Dataflow 管理者、Pub/Sub 編集者、Storage オブジェクト管理者、BigQuery データ編集者ロールに対して繰り返し、[保存] をクリックします。

      ロール付与の詳細については、コンソールを使用して IAM ロールを付与するをご覧ください。

gcloud

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. Install the Google Cloud CLI.

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

   gcloud init

4. Create or select 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.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

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

6. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs:

   gcloud services enable compute.googleapis.com dataflow.googleapis.com logging.googleapis.com bigquery.googleapis.com pubsub.googleapis.com storage.googleapis.com cloudresourcemanager.googleapis.com cloudscheduler.googleapis.com

7. If you're using a local shell, then create local authentication credentials for your user account:

   gcloud auth application-default login

   You don't need to do this if you're using Cloud Shell.

8. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

   gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.
9. Install the Google Cloud CLI.

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

    gcloud init

11. Create or select 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.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

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

13. Enable the Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler APIs:

    gcloud services enable compute.googleapis.com dataflow.googleapis.com logging.googleapis.com bigquery.googleapis.com pubsub.googleapis.com storage.googleapis.com cloudresourcemanager.googleapis.com cloudscheduler.googleapis.com

14. If you're using a local shell, then create local authentication credentials for your user account:

    gcloud auth application-default login

    You don't need to do this if you're using Cloud Shell.

15. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

    gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

    • Replace ROLE with each individual role.

16. Compute Engine のデフォルト サービス アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.admin
    • roles/pubsub.editor
    • roles/bigquery.dataEditor

    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

    次のように置き換えます。

    • PROJECT_ID: プロジェクト ID。
    • PROJECT_NUMBER: プロジェクトの番号。プロジェクト番号を確認するには、gcloud projects describe コマンドを使用します。
    • SERVICE_ACCOUNT_ROLE: 個々のロール。

サンプルソースとシンクを作成する

このセクションでは、次のものを作成します。

• Pub/Sub を使用したストリーミング データのソース
• BigQuery に読み込むデータのデータセット

Cloud Storage バケットを作成する

まず、Google Cloud コンソールまたは Google Cloud CLI を使用して Cloud Storage バケットを作成します。Dataflow パイプラインは、このバケットを一時ストレージの場所として使用します。

コンソール

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

   [バケット] に移動

2. [バケットを作成] をクリックします。

3. [バケットの作成] ページの [バケットに名前を付ける] で、バケット名の要件を満たす名前を入力します。Cloud Storage のバケット名は、グローバルに一意にする必要があります。他のオプションは選択しないでください。

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

gcloud

gcloud storage buckets create コマンドを実行します。

gcloud storage buckets create gs://BUCKET_NAME

BUCKET_NAME は、バケットの命名要件を満たす Cloud Storage バケットの名前に置き換えます。Cloud Storage のバケット名は、グローバルに一意にする必要があります。

Pub/Sub トピックとサブスクリプションを作成する

Pub/Sub トピックを作成し、そのトピックに対するサブスクリプションを作成します。

コンソール

トピックを作成するには、次の手順を行います。

1. Google Cloud コンソールで、Pub/Sub の [トピック] ページに移動します。

   [トピック] に移動

2. [トピックを作成] をクリックします。

3. [トピック ID] フィールドに、トピックの ID を入力します。トピックの指定方法については、トピックまたはサブスクリプションの指定方法のガイドラインをご覧ください。

4. [デフォルトのサブスクリプションを追加する] オプションは、そのまま保持します。他のオプションは選択しないでください。

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

gcloud

トピックを作成するには、gcloud pubsub topics create コマンドを実行します。サブスクリプションの指定方法については、トピックまたはサブスクリプションの指定方法のガイドラインをご覧ください。

gcloud pubsub topics create TOPIC_ID

TOPIC_ID は、Pub/Sub トピックの名前に置き換えます。

トピックのサブスクリプションを作成するには、gcloud pubsub subscriptions create コマンドを実行します。

gcloud pubsub subscriptions create --topic TOPIC_ID SUBSCRIPTION_ID

SUBSCRIPTION_ID は、Pub/Sub サブスクリプションの名前に置き換えます。

Cloud Scheduler ジョブを作成して実行する

2 つの Cloud Scheduler ジョブを作成して実行します。1 つは Pub/Sub トピックに高評価を公開するジョブ、もう 1 つは否定的な評価を公開するジョブです。

コンソール

高評価の Cloud Scheduler ジョブを作成します。

1. コンソールで [Cloud Scheduler] ページにアクセスします。

   Cloud Scheduler に移動

2. [ジョブの作成] ボタンをクリックします。

3. positive-ratings-publisher という名前を入力します。

4. このチュートリアルでコマンドを実行する場所に近接した Dataflow リージョンを選択します。REGION 変数の値は有効なリージョン名にする必要があります。リージョンとロケーションの詳細については、Dataflow のロケーションをご覧ください。

5. ジョブの [頻度] を unix-cron 形式 * * * * * で指定します。

   詳細については、cron ジョブ スケジュールの構成をご覧ください。

6. タイムゾーンを選択します。

7. [続行] をクリックします。

8. [ターゲット] リストで、Pub/Sub を選択します。

9. リストからトピック名を選択します。

10. ターゲットに送信する次のメッセージ文字列を追加します。 {"url": "https://beam.apache.org/", "review": "positive"}

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

これで、1 分ごとに Pub/Sub トピックに肯定的な評価のメッセージを送信する cron ジョブが作成されました。Cloud Functions の関数はこのトピックをサブスクライブします。

否定的な評価の Cloud Scheduler ジョブを作成します。

1. コンソールの [Cloud Scheduler] ページで、[ジョブを作成] ボタンをクリックします。

2. negative-ratings-publisher という名前を入力します。

3. ジョブを実行するリージョンを選択します。

4. ジョブの [頻度] を unix-cron 形式 */2 * * * * で指定します。

   詳細については、cron ジョブ スケジュールの構成をご覧ください。

5. タイムゾーンを選択します。

6. [続行] をクリックします。

7. [ターゲット] リストで、Pub/Sub を選択します。

8. リストからトピック名を選択します。

9. ターゲットに送信する次のメッセージ文字列を追加します。 {"url": "https://beam.apache.org/", "review": "negative"}

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

これで、否定的な評価のメッセージを Pub/Sub トピックに 2 分ごとに送信する cron ジョブが作成されました。Cloud Functions の関数はこのトピックをサブスクライブします。

gcloud

1. このチュートリアルの Cloud Scheduler ジョブを作成するには、gcloud scheduler jobs create コマンドを使用します。このステップでは、1 分間に 1 つのメッセージを公開する肯定的な評価のパブリッシャーが作成されます。

   gcloud scheduler jobs create pubsub positive-ratings-publisher \
     --schedule="* * * * *" \
     --location=DATAFLOW_REGION \
     --topic="TOPIC_ID" \
     --message-body='{"url": "https://beam.apache.org/", "review": "positive"}'
   

   DATAFLOW_REGION は、Dataflow ジョブをデプロイするリージョンに置き換えます。このチュートリアルでコマンドを実行する場所に近い Dataflow リージョンを選択します。REGION 変数の値は有効なリージョン名にする必要があります。

2. Cloud Scheduler ジョブを開始するには、gcloud scheduler jobs run コマンドを使用します。

   gcloud scheduler jobs run --location=DATAFLOW_REGION positive-ratings-publisher
   

3. 2 分ごとに 1 つのメッセージを公開する否定的な評価のパブリッシャーを作成して実行します。このステップでは、2 分ごとに 1 つのメッセージを公開する否定的な評価のパブリッシャーが作成されます。

   gcloud scheduler jobs create pubsub negative-ratings-publisher \
     --schedule="*/2 * * * *" \
     --location=DATAFLOW_REGION  \
     --topic="TOPIC_ID" \
     --message-body='{"url": "https://beam.apache.org/", "review": "negative"}'
   

4. 2 番目の Cloud Scheduler ジョブを開始します。

   gcloud scheduler jobs run --location=DATAFLOW_REGION negative-ratings-publisher
   

BigQuery データセットを作成する

Pub/Sub トピックに適したスキーマを使用して、BigQuery のデータセットとテーブルを作成します。

コンソール

BigQuery データセットを作成します。

1. Google Cloud コンソールで [BigQuery] ページを開きます。

   [BigQuery] ページに移動

2. [エクスプローラ] パネルで、データセットを作成するプロジェクトを選択します。

3. アクション オプションを開いて、[データセットを作成] をクリックします。

4. [データセットを作成] ページで次の操作を行います。

   • [データセット ID] に「tutorial_dataset」と入力します。

   • [データのロケーション] で、データセットの地理的位置を選択します。データセットの作成後はロケーションを変更できません。

     注: データセットのロケーションとして EU または EU にあるリージョンを選択した場合、Core Dataflow Customer Data は EU に保存されます。Core Dataflow Customer Data はサービス固有の規約で定義されています。

   • 他のオプションは選択しないでください。

   • [データセットを作成] をクリックします。

スキーマを使用して BigQuery テーブルを作成します。

1. [エクスプローラ] パネルでプロジェクトを開いて、tutorial_dataset データセットを選択します。

2. アクション オプションを開いて、[開く] をクリックします。

3. 詳細パネルで [テーブルを作成] をクリックします。

4. [テーブルの作成] ページの [ソース] セクションで、[空のテーブル] を選択します。

5. [テーブルの作成] ページの [送信先] セクションで、次の操作を行います。

   • [データセット名] が tutorial_dataset に設定されていることを確認します。
   • [テーブル名] フィールドに「tutorial」と入力します。
   • [テーブルタイプ] が [ネイティブ テーブル] に設定されていることを確認します。

6. [スキーマ] セクションにスキーマ定義を入力します。[テキストとして編集] を有効にし、以下のテーブル スキーマを JSON 配列として入力します。

   [
     {
       "mode": "NULLABLE",
       "name": "url",
       "type": "STRING"
     },
     {
       "mode": "NULLABLE",
       "name": "review",
       "type": "STRING"
     }
   ]
   

7. [パーティションとクラスタの設定] はデフォルト値（No partitioning）のままにします。

8. [詳細オプション] セクションの [暗号化] の値は、デフォルト（Google-managed key）のままにします。Dataflow では、デフォルトの処理として保存されているお客様のコンテンツを暗号化しています。

9. [テーブルを作成] をクリックします。

gcloud

bq mk コマンドを使用して、データセットを作成します。

bq --location=DATAFLOW_REGION mk \
PROJECT_ID:tutorial_dataset

PROJECT_ID は、プロジェクトのプロジェクト ID に置き換えます。

データセットにテーブルを作成するには、--table フラグまたは -t フラグを指定した bq mk コマンドを使用します。

bq mk \
    --table \
    PROJECT_ID:tutorial_dataset.tutorial \
    url:STRING,review:STRING

ユーザー定義関数（UDF）を作成する

JavaScript UDF を作成して、Google 提供の Pub/Sub Subscription to BigQuery テンプレートを拡張することもできます。UDF を使用すると、現在テンプレートにないデータ変換を定義し、テンプレートに挿入できます。

注: Dataflow UDF は JavaScript で記述され、Nashorn JavaScript エンジンで実行可能であることが必要です。

次の UDF は、受信する評価の URL を検証します。URL のない評価や URL が正しくない評価は、同じプロジェクトとデータセット内で _error_records のサフィックスを持つ別の出力テーブル（デッドレター テーブルとも呼ばれます）に転送されます。

JavaScript

/**
 * User-defined function (UDF) to transform events
 * as part of a Dataflow template job.
 *
 * @param {string} inJson input Pub/Sub JSON message (stringified)
 */
 function process(inJson) {
    const obj = JSON.parse(inJson);
    const includePubsubMessage = obj.data && obj.attributes;
    const data = includePubsubMessage ? obj.data : obj;

    if (!data.hasOwnProperty('url')) {
      throw new Error("No url found");
    } else if (data.url !== "https://beam.apache.org/") {
      throw new Error("Unrecognized url");
    }

    return JSON.stringify(obj);
  }

この JavaScript スニペットを、先ほど作成した Cloud Storage バケットに保存します。

パイプラインを実行する

Google 提供の Pub/Sub Subscription to BigQuery テンプレートを使用して、ストリーミング パイプラインを実行します。このパイプラインは Pub/Sub トピックから受信データを取得し、BigQuery データセットに出力します。

コンソール

1. Google Cloud コンソールで、Dataflow の [ジョブ] ページに移動します。

   [ジョブ] に移動

2. [テンプレートからジョブを作成] をクリックします。

3. Dataflow ジョブの [ジョブ名] を入力します。

4. [リージョン エンドポイント] で、Dataflow ジョブのリージョンを選択します。

5. [Dataflow テンプレート] で、Pub/Sub Subscription to BigQuery テンプレートを選択します。

6. [BigQuery output table] に、次のテキストを入力します。

   PROJECT_ID:tutorial_dataset.tutorial
   

7. [Pub/Sub input subscription] に、次のように入力します。

   projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
   

   PROJECT_ID は、BigQuery データセットを作成したプロジェクトのプロジェクト ID に置き換えます。SUBSCRIPTION_ID は、Pub/Sub サブスクリプションの名前に置き換えます。

8. [一時的な場所] に、次のように入力します。

   gs://BUCKET_NAME/temp/
   

   BUCKET_NAME を Cloud Storage バケットの名前に置き換えます。temp フォルダには、ステージングされたパイプライン ジョブなどの一時ファイルが格納されます。

9. 省略可: ジョブに UDF を含めるには、[オプション パラメータ] を開きます。

   1. [JavaScript UDF path in Cloud Storage] に、次のように入力します。

      gs://BUCKET_NAME/dataflow_udf_transform.js
      

   2. [JavaScript UDF name] に、次のように入力します。

      process
      

10. [ジョブを実行] をクリックします。

テンプレートがメッセージをデッドレター テーブルに転送できるかどうかを確認するには、URL のない評価、または誤った URL を含む評価を公開します。

1. Pub/Sub の [トピック] ページに移動します。

2. TOPIC_ID をクリックします。

3. [メッセージ] に移動します。

4. [メッセージをパブリッシュ] をクリックします。

5. [メッセージ本文] に、URL がない評価または誤った URL を含む評価を入力します。次に例を示します。

   {"url": "https://beam.apache.org/documentation/sdks/java/", "review": "positive"}
   

6. [公開] をクリックします。

gcloud

注: Google Cloud CLI を使用してクラシック テンプレートを実行するには、gcloud CLI バージョン 138.0.0 以降が必要です。

シェルまたはターミナルでテンプレートを実行するには、gcloud dataflow jobs run コマンドを使用します。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-DATAFLOW_REGION/latest/PubSub_Subscription_to_BigQuery \
    --region DATAFLOW_REGION \
    --staging-location gs://BUCKET_NAME/temp \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID,\
outputTableSpec=PROJECT_ID:tutorial_dataset.tutorial

JOB_NAME は、一意の名前で置き換えます。

必要に応じて、UDF でテンプレートを実行するには、次のコマンドを使用します。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-DATAFLOW_REGION/latest/PubSub_Subscription_to_BigQuery \
    --region DATAFLOW_REGION \
    --staging-location gs://BUCKET_NAME/temp \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID,\
outputTableSpec=PROJECT_ID:tutorial_dataset.tutorial,\
javascriptTextTransformGcsPath=gs://BUCKET_NAME/dataflow_udf_transform.js,\
javascriptTextTransformFunctionName=process

テンプレートがメッセージをデッドレター テーブルに転送できるかどうかを確認するには、URL のない評価、または誤った URL を含む評価を公開します。例:

gcloud pubsub topics publish TOPIC_ID \
  --message='{"url": "https://beam.apache.org/documentation/sdks/java/", "review": "positive"}'

結果を表示する

BigQuery テーブルに書き込まれたデータを表示します。

コンソール

1. Google Cloud コンソールで [BigQuery] ページに移動します。
   [BigQuery] ページに移動

2. [クエリエディタ] ペインで、次のクエリを実行します。

   SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial`
   LIMIT 1000
   

   テーブルにデータが表示されるまで、最大で 1 分かかることがあります。

   クエリは、過去 24 時間以内にテーブルに追加された行を返します。標準 SQL を使用してクエリを実行することもできます。

   デッドレター テーブルに一部のエラーレコードが書き込まれることが予想される場合は、クエリ内でテーブル名 tutorial_error_records を使用します。例:

   SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial_error_records`
   LIMIT 1000
   

gcloud

次のクエリを実行して、BigQuery の結果を確認します。

bq query --use_legacy_sql=false 'SELECT * FROM `'"PROJECT_ID.tutorial_dataset.tutorial"'`'

このパイプラインの実行中は、1 分おきに BigQuery テーブルに新しい行が追加されます。

デッドレター テーブルに一部のエラーレコードが書き込まれることが予想される場合は、クエリ内でテーブル名 tutorial_error_records を使用します。次に例を示します。

SELECT * FROM `PROJECT_ID.tutorial_dataset.tutorial_error_records`
LIMIT 1000

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、リソースを含むプロジェクトを削除するか、プロジェクトを維持して個々のリソースを削除します。

プロジェクトを削除する

課金を停止する最も簡単な方法は、チュートリアル用に作成した Google Cloud プロジェクトを削除することです。

コンソール

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

• プロジェクト内のすべてのものが削除されます。このドキュメントのタスクで既存のプロジェクトを使用した場合、それを削除すると、そのプロジェクトで行った他の作業もすべて削除されます。
• カスタム プロジェクト 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.

gcloud

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

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

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

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

個々のリソースを削除する

後でプロジェクトを再利用する場合は、プロジェクトを残したまま、チュートリアル中に作成したリソースを削除します。

Dataflow パイプラインを停止する

コンソール

1. Google Cloud コンソールで、Dataflow の [ジョブ] ページに移動します。

   [ジョブ] に移動

2. 停止するジョブをクリックします。

   ジョブを停止するには、ジョブのステータスが「実行中」でなければなりません。

3. ジョブの詳細ページで、[停止] をクリックします。

4. [キャンセル] をクリックします。

5. 選択を確定するには、[ジョブの停止] をクリックします。

gcloud

Dataflow ジョブをキャンセルするには、gcloud dataflow jobs コマンドを使用します。

gcloud dataflow jobs list \
  --filter 'NAME=JOB_NAME AND STATE=Running' \
  --format 'value(JOB_ID)' \
  --region "DATAFLOW_REGION" \
  | xargs gcloud dataflow jobs cancel --region "DATAFLOW_REGION"

Google Cloud プロジェクトのリソースをクリーンアップする

コンソール

1. Cloud Scheduler ジョブを削除します。

   1. Google Cloud コンソールの [Cloud Scheduler] ページに移動します。

      Cloud Scheduler に移動

   2. ジョブを選択します。

   3. ページの上部にある [削除] ボタンをクリックして、削除操作を確定します。

2. Pub/Sub トピックとサブスクリプションを削除します。

   1. Google Cloud コンソールで Pub/Sub の [トピック] ページに移動します。

      [トピック] に移動

   2. 作成したトピックを選択します。

   3. [削除] をクリックして、トピックを永続的に削除します。

   4. Google Cloud コンソールで Pub/Sub の [サブスクリプション] ページに移動します。

      [サブスクリプション] に移動

   5. トピックで作成したサブスクリプションを選択します。

   6. [削除] をクリックして、サブスクリプションを永続的に削除します。

3. BigQuery テーブルとデータセットを削除します。

   1. Google Cloud コンソールで [BigQuery] ページに移動します。

      [BigQuery] に移動

   2. [エクスプローラ] パネルで、プロジェクトを展開します。

   3. 削除するデータセットの横にある「アクションを表示」アイコン をクリックし、[削除] をクリックします。

4. Cloud Storage バケットを削除します。

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

      [バケット] に移動

   2. 削除するバケットを選択し、[削除] をクリックして、指示に沿って操作します。

gcloud

1. Cloud Scheduler のジョブを削除するには、gcloud scheduler jobs delete コマンドを使用します。

   gcloud scheduler jobs delete negative-ratings-publisher --location=DATAFLOW_REGION
   

   gcloud scheduler jobs delete positive-ratings-publisher --location=DATAFLOW_REGION
   

2. Pub/Sub サブスクリプションとトピックを削除するには、gcloud pubsub subscriptions delete コマンドと gcloud pubsub topics delete コマンドを使用します。

   gcloud pubsub subscriptions delete SUBSCRIPTION_ID
   gcloud pubsub topics delete TOPIC_ID
   

3. BigQuery テーブルを削除するには、bq rm コマンドを使用します。

   bq rm -f -t PROJECT_ID:tutorial_dataset.tutorial
   

4. BigQuery データセットを削除します。データセットだけで料金は発生しません。

   注意: 次のコマンドを実行すると、データセット内のすべてのテーブルも削除されます。テーブルとデータは復元できません。

   bq rm -r -f -d PROJECT_ID:tutorial_dataset
   

5. Cloud Storage バケットを削除するには、gcloud storage rm コマンドを使用します。バケットだけでは料金は発生しません。

   注意: 次のコマンドを実行すると、バケット内のすべてのオブジェクトも削除されます。これらのオブジェクトは復元できません。

   gcloud storage rm gs://BUCKET_NAME --recursive
   

認証情報を取り消す

コンソール

プロジェクトを残しておく場合は、Compute Engine のデフォルトのサービス アカウントに付与したロールを取り消します。

1. Google Cloud コンソールの [IAM] ページに移動します。

[IAM] に移動

1. プロジェクト、フォルダ、または組織を選択します。

2. アクセス権を取り消すプリンシパルを含む行を探します。その行の「プリンシパルを編集します」アイコン をクリックします。

3. 取り消すロールごとに [削除] ボタンをクリックして、[保存] をクリックします。

gcloud

• プロジェクトを残しておく場合は、Compute Engine のデフォルトのサービス アカウントに付与したロールを取り消します。以下の IAM ロールごとに次のコマンドを 1 回実行します。
  • roles/dataflow.admin
  • roles/dataflow.worker
  • roles/storage.admin
  • roles/pubsub.editor
  • roles/bigquery.dataEditor

    gcloud projects remove-iam-policy-binding <var>PROJECT_ID</var> \
    --member=serviceAccount:<var>PROJECT_NUMBER</var>[e-mail] \
    --role=<var>ROLE</var>
  

• Optional: Revoke the authentication credentials that you created, and delete the local credential file.

  gcloud auth application-default revoke

• Optional: Revoke credentials from the gcloud CLI.

  gcloud auth revoke

次のステップ

• UDF によって Dataflow テンプレートを拡張する。
• Dataflow テンプレートの使用方法を学習する。
• Google 提供のすべてのテンプレートを表示する。
• Pub/Sub を使用してトピックを作成して使用する方法と、pull サブスクリプションを作成する方法を確認する。
• Cloud Scheduler を使用して cron ジョブをスケジュールして実行する方法を学習する。
• BigQuery を使用してデータセットを作成する方法を確認する。
• Pub/Sub サブスクリプションについて学習する。
• Google Cloud に関するリファレンス アーキテクチャ、図、ベスト プラクティスを確認する。Cloud アーキテクチャ センターをご覧ください。