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

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

このワークフローのメリットは、UDF を使用して Google 提供のストリーミングテンプレートを拡張できることです。Pub/Sub からデータを pull して BigQuery に出力する必要があるが、テンプレートを拡張する必要がない場合は、より簡単なワークフローで Pub/Sub から BigQuery へのサブスクリプション機能を使用できます。

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

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

データ変換を提供するために Dataflow を構成することなくメッセージを直接 BigQuery に書き込む場合は、Pub/Sub to BigQuery subscription 機能を使用します。

目標

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

費用

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

Dataflow
Pub/Sub
Cloud Storage
Cloud Scheduler
BigQuery

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

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

始める前に

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

コンソール

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

Google Cloud Console の [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

プロジェクトセレクタに移動

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler API を有効にします。

API を有効にする

Google Cloud Console の [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

プロジェクトセレクタに移動

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler API を有効にします。

API を有効にする

このチュートリアルの手順を完了するには、ユーザーアカウントにサービスアカウントユーザーのロールが必要です。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

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

gcloud CLI を初期化するには:

gcloud init

Google Cloud プロジェクトを作成または選択します。

Google Cloud プロジェクトを作成します。
```
gcloud projects create PROJECT_ID
```
PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。
作成した Google Cloud プロジェクトを選択します。
```
gcloud config set project PROJECT_ID
```
PROJECT_ID は、実際の Google Cloud プロジェクト名に置き換えます。

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler API を有効にします。

gcloud services enable compute.googleapis.comdataflow.googleapis.comlogging.googleapis.combigquery.googleapis.compubsub.googleapis.comstorage.googleapis.comcloudresourcemanager.googleapis.comcloudscheduler.googleapis.com

Google アカウントのローカル認証情報を作成します。

gcloud auth application-default login

Google アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。 roles/iam.serviceAccountUser

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

PROJECT_ID は、実際のプロジェクト ID に置き換えます。
EMAIL_ADDRESS は実際のメールアドレスに置き換えます。
ROLE は、個々のロールに置き換えます。

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

gcloud CLI を初期化するには:

gcloud init

Google Cloud プロジェクトを作成または選択します。

Google Cloud プロジェクトを作成します。
```
gcloud projects create PROJECT_ID
```
PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。
作成した Google Cloud プロジェクトを選択します。
```
gcloud config set project PROJECT_ID
```
PROJECT_ID は、実際の Google Cloud プロジェクト名に置き換えます。

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Compute Engine, Dataflow, Cloud Logging, BigQuery, Pub/Sub, Cloud Storage, Resource Manager, and Cloud Scheduler API を有効にします。

gcloud services enable compute.googleapis.comdataflow.googleapis.comlogging.googleapis.combigquery.googleapis.compubsub.googleapis.comstorage.googleapis.comcloudresourcemanager.googleapis.comcloudscheduler.googleapis.com

Google アカウントのローカル認証情報を作成します。

gcloud auth application-default login

Google アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。 roles/iam.serviceAccountUser

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

PROJECT_ID は、実際のプロジェクト ID に置き換えます。
EMAIL_ADDRESS は実際のメールアドレスに置き換えます。
ROLE は、個々のロールに置き換えます。

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:PROJECT_NUMBER-compute@developer.gserviceaccount.com" --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 パイプラインは、このバケットを一時ストレージの場所として使用します。

コンソール

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

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

gcloud

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

gcloud storage buckets create gs://BUCKET_NAME

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

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

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

コンソール

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

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

[トピック] に移動
[トピックを作成] をクリックします。
[トピック ID] フィールドに、トピックの ID を入力します。トピックの指定方法については、トピックまたはサブスクリプションの指定方法のガイドラインをご覧ください。
[デフォルトのサブスクリプションを追加する] オプションは、そのまま保持します。他のオプションは選択しないでください。
[トピックを作成] をクリックします。

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 ジョブを作成します。

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

Cloud Scheduler に移動
[ジョブの作成] ボタンをクリックします。
positive-ratings-publisher という名前を入力します。
このチュートリアルでコマンドを実行する場所に近接した Dataflow リージョンを選択します。REGION 変数の値は有効なリージョン名にする必要があります。リージョンとロケーションの詳細については、Dataflow のロケーションをご覧ください。
ジョブの [頻度] を unix-cron 形式 * * * * * で指定します。

詳細については、cron ジョブスケジュールの構成をご覧ください。
タイムゾーンを選択します。
[続行] をクリックします。
[ターゲット] リストで、Pub/Sub を選択します。
リストからトピック名を選択します。
ターゲットに送信する次のメッセージ文字列を追加します。 {"url": "https://beam.apache.org/", "review": "positive"}
[作成] をクリックします。

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

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

コンソールの [Cloud Scheduler] ページで、[ジョブを作成] ボタンをクリックします。
negative-ratings-publisher という名前を入力します。
ジョブを実行するリージョンを選択します。
ジョブの [頻度] を unix-cron 形式 */2 * * * * で指定します。

詳細については、cron ジョブスケジュールの構成をご覧ください。
タイムゾーンを選択します。
[続行] をクリックします。
[ターゲット] リストで、Pub/Sub を選択します。
リストからトピック名を選択します。
ターゲットに送信する次のメッセージ文字列を追加します。 {"url": "https://beam.apache.org/", "review": "negative"}
[作成] をクリックします。

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

gcloud

このチュートリアルの 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 変数の値は有効なリージョン名にする必要があります。
Cloud Scheduler ジョブを開始するには、gcloud scheduler jobs run コマンドを使用します。
```
gcloud scheduler jobs run --location=DATAFLOW_REGION positive-ratings-publisher
```
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"}'
```

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

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

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

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

コンソール

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

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

[BigQuery] ページに移動
[エクスプローラ] パネルで、データセットを作成するプロジェクトを選択します。
アクション オプションを開いて、[データセットを作成] をクリックします。
[データセットを作成] ページで次の操作を行います。
- [データセット ID] に「tutorial_dataset」と入力します。
- [データのロケーション] で、データセットの地理的位置を選択します。データセットの作成後はロケーションを変更できません。
  
  注: データセットのロケーションとして EU または EU にあるリージョンを選択した場合、Core Dataflow Customer Data は EU に保存されます。Core Dataflow Customer Data はサービス固有の規約で定義されています。
- 他のオプションは選択しないでください。
- [データセットを作成] をクリックします。

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

[エクスプローラ] パネルでプロジェクトを開いて、tutorial_dataset データセットを選択します。
アクション オプションを開いて、[開く] をクリックします。
詳細パネルで [テーブルを作成] をクリックします。
[テーブルの作成] ページの [ソース] セクションで、[空のテーブル] を選択します。
[テーブルの作成] ページの [送信先] セクションで、次の操作を行います。
- [データセット名] が tutorial_dataset に設定されていることを確認します。
- [テーブル名] フィールドに「tutorial」と入力します。
- [テーブルタイプ] が [ネイティブテーブル] に設定されていることを確認します。
[スキーマ] セクションにスキーマ定義を入力します。[テキストとして編集] を有効にし、以下のテーブルスキーマを JSON 配列として入力します。
```
[
  {
    "mode": "NULLABLE",
    "name": "url",
    "type": "STRING"
  },
  {
    "mode": "NULLABLE",
    "name": "review",
    "type": "STRING"
  }
]
```
[パーティションとクラスタの設定] はデフォルト値（No partitioning）のままにします。
[詳細オプション] セクションの [暗号化] の値は、デフォルト（Google-managed key）のままにします。Dataflow では、デフォルトの処理として保存されているお客様のコンテンツを暗号化しています。
[テーブルを作成] をクリックします。

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 を使用すると、現在テンプレートにないデータ変換を定義し、テンプレートに挿入できます。

次の 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 データセットに出力します。

コンソール

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

[ジョブ] に移動
[テンプレートからジョブを作成] をクリックします。
Dataflow ジョブの [ジョブ名] を入力します。
[リージョンエンドポイント] で、Dataflow ジョブのリージョンを選択します。
[Dataflow テンプレート] で、Pub/Sub Subscription to BigQuery テンプレートを選択します。
[BigQuery output table] に、次のテキストを入力します。
```
PROJECT_ID:tutorial_dataset.tutorial
```
[Pub/Sub input subscription] に、次のように入力します。
```
projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
```
PROJECT_ID は、BigQuery データセットを作成したプロジェクトのプロジェクト ID に置き換えます。SUBSCRIPTION_ID は、Pub/Sub サブスクリプションの名前に置き換えます。
[一時的な場所] に、次のように入力します。
```
gs://BUCKET_NAME/temp/
```
BUCKET_NAME を Cloud Storage バケットの名前に置き換えます。temp フォルダには、ステージングされたパイプラインジョブなどの一時ファイルが格納されます。
省略可: ジョブに UDF を含めるには、[オプションパラメータ] を開きます。
1. [JavaScript UDF path in Cloud Storage] に、次のように入力します。
```
gs://BUCKET_NAME/dataflow_udf_transform.js
```
2. [JavaScript UDF name] に、次のように入力します。
```
process
```
[ジョブを実行] をクリックします。

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

Pub/Sub の [トピック] ページに移動します。
TOPIC_ID をクリックします。
[メッセージ] に移動します。
[メッセージをパブリッシュ] をクリックします。
[メッセージ本文] に、URL がない評価または誤った URL を含む評価を入力します。次に例を示します。
```
{"url": "https://beam.apache.org/documentation/sdks/java/", "review": "positive"}
```
[公開] をクリックします。

gcloud

シェルまたはターミナルでテンプレートを実行するには、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 テーブルに書き込まれたデータを表示します。

コンソール

Google Cloud コンソールで [BigQuery] ページに移動します。
[BigQuery] ページに移動
[クエリエディタ] ペインで、次のクエリを実行します。
```
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）を保持するには、プロジェクト全体ではなくプロジェクト内の選択したリソースだけを削除します。

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

Google Cloud コンソールで、[リソースの管理] ページに移動します。
[リソースの管理] に移動
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

gcloud

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

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

Google Cloud プロジェクトを削除します。

gcloud projects delete PROJECT_ID

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

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

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

コンソール

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

[ジョブ] に移動
停止するジョブをクリックします。

ジョブを停止するには、ジョブのステータスが「実行中」でなければなりません。
ジョブの詳細ページで、[停止] をクリックします。
[キャンセル] をクリックします。
選択を確定するには、[ジョブの停止] をクリックします。

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 プロジェクトのリソースをクリーンアップする

コンソール

Cloud Scheduler ジョブを削除します。
1. Google Cloud コンソールの [Cloud Scheduler] ページに移動します。
  
  Cloud Scheduler に移動
2. ジョブを選択します。
3. ページの上部にある [削除] ボタンをクリックして、削除操作を確定します。
Pub/Sub トピックとサブスクリプションを削除します。
1. Google Cloud コンソールで Pub/Sub の [トピック] ページに移動します。
  
  [トピック] に移動
2. 作成したトピックを選択します。
3. [削除] をクリックして、トピックを永続的に削除します。
4. Google Cloud コンソールで Pub/Sub の [サブスクリプション] ページに移動します。
  
  [サブスクリプション] に移動
5. トピックで作成したサブスクリプションを選択します。
6. [削除] をクリックして、サブスクリプションを永続的に削除します。
BigQuery テーブルとデータセットを削除します。
1. Google Cloud コンソールで [BigQuery] ページに移動します。
  
  [BigQuery] に移動
2. [エクスプローラ] パネルで、プロジェクトを展開します。
3. 削除するデータセットの横にある「アクションを表示」アイコンをクリックし、[削除] をクリックします。
Cloud Storage バケットを削除します。
1. Google Cloud コンソールで、Cloud Storage の [バケット] ページに移動します。
  
  [バケット] に移動
2. 削除するバケットを選択し、[削除] をクリックして、指示に沿って操作します。

gcloud

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

Pub/Sub サブスクリプションとトピックを削除するには、gcloud pubsub subscriptions delete コマンドと gcloud pubsub topics delete コマンドを使用します。
```
gcloud pubsub subscriptions delete SUBSCRIPTION_ID
gcloud pubsub topics delete TOPIC_ID
```
BigQuery テーブルを削除するには、bq rm コマンドを使用します。
```
bq rm -f -t PROJECT_ID:tutorial_dataset.tutorial
```
BigQuery データセットを削除します。データセットだけで料金は発生しません。

注意: 次のコマンドを実行すると、データセット内のすべてのテーブルも削除されます。テーブルとデータは復元できません。
```
bq rm -r -f -d PROJECT_ID:tutorial_dataset
```
Cloud Storage バケットを削除するには、gcloud storage rm コマンドを使用します。バケットだけでは料金は発生しません。

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

認証情報を取り消す

コンソール

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

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

[IAM] に移動

プロジェクト、フォルダ、または組織を選択します。
アクセス権を取り消すプリンシパルを含む行を探します。その行の「プリンシパルを編集します」アイコンをクリックします。
取り消すロールごとに [削除] ボタンをクリックして、[保存] をクリックします。

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>-compute@developer.gserviceaccount.com \
  --role=<var>ROLE</var>
```

作成した認証情報を取り消して、ローカル認証情報ファイルを削除します。
```
gcloud auth application-default revoke
```

（省略可）gcloud CLI から認証情報を取り消します。
```
gcloud auth revoke
```

次のステップ

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