Dataflow SQL によるストリーミング データの結合

このチュートリアルでは、Dataflow SQL を使用して Pub/Sub のデータ ストリームを BigQuery テーブルのデータと結合する方法について説明します。

目標

このチュートリアルの内容は次のとおりです。

• Pub/Sub ストリーミング データと BigQuery テーブルデータを結合する Dataflow SQL クエリを作成します。
• Dataflow SQL UI から Dataflow ジョブをデプロイします。

費用

このチュートリアルでは、課金対象である次の Google Cloud コンポーネントを使用します。

• Dataflow
• Cloud Storage
• Pub/Sub
• Data Catalog

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

始める前に

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

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

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、新しいプロジェクトを作成してください。チュートリアルの終了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

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

3. Cloud プロジェクトに対して課金が有効になっていることを確認します。プロジェクトに対して課金が有効になっていることを確認する方法を学習する。

4. Cloud Dataflow、Compute Engine、Logging、Cloud Storage、Cloud Storage JSON、BigQuery、Cloud Pub/Sub、Cloud Resource Manager、Data Catalog。 API を有効にします。

   API を有効にする

5. サービス アカウントを作成します。

   1. Cloud Console で [サービス アカウントの作成] ページに移動します。

      [サービス アカウントの作成] に移動
   2. プロジェクトを選択します。

   3. [サービス アカウント名] フィールドに名前を入力します。Cloud Console は、この名前に基づいて [サービス アカウント ID] フィールドに入力します。

      [サービス アカウントの説明] フィールドに説明を入力します。例: Service account for quickstart

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

   5. [ロールを選択] フィールドをクリックします。

      [クイック アクセス] で [基本]、[オーナー] の順にクリックします。

      注: [ロール] フィールドは、サービス アカウントがプロジェクト内のどのリソースにアクセスできるかに影響します。これらのロールは後で取り消すことも、追加のロールを付与することもできます。本番環境では、オーナー、編集者、閲覧者のロールを付与しないでください。代わりに、ニーズに合わせて事前定義された役割またはカスタムロールを付与します。
   6. [続行] をクリックします。

   7. [完了] をクリックして、サービス アカウントの作成を完了します。

      ブラウザ ウィンドウは閉じないでください。次のステップでこれを使用します。

6. サービス アカウント キーを作成します。

   1. Cloud Console で、作成したサービス アカウントのメールアドレスをクリックします。
   2. [キー] をクリックします。
   3. [鍵を追加]、[新しい鍵を作成] の順にクリックします。
   4. [作成] をクリックします。JSON キーファイルがパソコンにダウンロードされます。
   5. [閉じる] をクリックします。

7. 環境変数 GOOGLE_APPLICATION_CREDENTIALS を、サービス アカウント キーが含まれる JSON ファイルのパスに設定します。 この変数は現在のシェル セッションにのみ適用されるため、新しいセッションを開く場合は、変数を再度設定します。

   例: Linux または macOS

   export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

   例:

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   例: Windows

   PowerShell の場合:

   $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

   例:

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

   コマンド プロンプトの場合:

   set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

   KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

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

   注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、新しいプロジェクトを作成してください。チュートリアルの終了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

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

9. Cloud プロジェクトに対して課金が有効になっていることを確認します。プロジェクトに対して課金が有効になっていることを確認する方法を学習する。

10. Cloud Dataflow、Compute Engine、Logging、Cloud Storage、Cloud Storage JSON、BigQuery、Cloud Pub/Sub、Cloud Resource Manager、Data Catalog。 API を有効にします。

    API を有効にする

11. サービス アカウントを作成します。

    1. Cloud Console で [サービス アカウントの作成] ページに移動します。

       [サービス アカウントの作成] に移動
    2. プロジェクトを選択します。

    3. [サービス アカウント名] フィールドに名前を入力します。Cloud Console は、この名前に基づいて [サービス アカウント ID] フィールドに入力します。

       [サービス アカウントの説明] フィールドに説明を入力します。例: Service account for quickstart

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

    5. [ロールを選択] フィールドをクリックします。

       [クイック アクセス] で [基本]、[オーナー] の順にクリックします。

       注: [ロール] フィールドは、サービス アカウントがプロジェクト内のどのリソースにアクセスできるかに影響します。これらのロールは後で取り消すことも、追加のロールを付与することもできます。本番環境では、オーナー、編集者、閲覧者のロールを付与しないでください。代わりに、ニーズに合わせて事前定義された役割またはカスタムロールを付与します。
    6. [続行] をクリックします。

    7. [完了] をクリックして、サービス アカウントの作成を完了します。

       ブラウザ ウィンドウは閉じないでください。次のステップでこれを使用します。

12. サービス アカウント キーを作成します。

    1. Cloud Console で、作成したサービス アカウントのメールアドレスをクリックします。
    2. [キー] をクリックします。
    3. [鍵を追加]、[新しい鍵を作成] の順にクリックします。
    4. [作成] をクリックします。JSON キーファイルがパソコンにダウンロードされます。
    5. [閉じる] をクリックします。

13. 環境変数 GOOGLE_APPLICATION_CREDENTIALS を、サービス アカウント キーが含まれる JSON ファイルのパスに設定します。 この変数は現在のシェル セッションにのみ適用されるため、新しいセッションを開く場合は、変数を再度設定します。

    例: Linux または macOS

    export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

    KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

    例:

    export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

    例: Windows

    PowerShell の場合:

    $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

    KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

    例:

    $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

    コマンド プロンプトの場合:

    set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

    KEY_PATH をサービス アカウント キーが含まれる JSON ファイルのパスに置き換えます。

14. Cloud SDK をインストールし、初期化します。インストール オプションを、次の中から選択します。場合によっては、このチュートリアルで使用するプロジェクトに project プロパティを設定する必要があります。
15. Cloud Console で Dataflow SQL ウェブ UI に移動します。最後にアクセスしたプロジェクトが開きます。別のプロジェクトに切り替えるには、Dataflow SQL ウェブ UI の先頭に表示されているプロジェクト名をクリックし、使用するプロジェクトを検索します。
    Dataflow SQL ウェブ UI に移動
    

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

サンプルで使用するソースを作成します。このチュートリアルのサンプルでは、次のソースを使用します。

• transactions という名前の Pub/Sub トピック - Pub/Sub トピックのサブスクリプションを介して送信されるトランザクション データのストリーム。各トランザクションのデータには、購入した商品、販売価格、購入場所などの情報が含まれています。Pub/Sub トピックを作成したら、トピックにメッセージを公開するスクリプトを作成します。このチュートリアルの後半で、このスクリプトを実行します。
• us_state_salesregions という名前の BigQuery テーブル - 州と販売地域をマッピングするテーブル。このテーブルを作成する前に、BigQuery データセットを作成する必要があります。

Pub/Sub トピックと publisher スクリプトを作成する

1. gcloud コマンドライン ツールを使用して、Pub/Sub トピックを作成します。トピックには、transactions という名前を付けます。

   gcloud pubsub topics create transactions

2. Pub/Sub トピックにメッセージを公開する Python スクリプトを作成します。まだインストールしていない場合は Python をインストールする必要があります。このスクリプトは、SQL クエリを実行する直前にコマンドライン ウィンドウで実行します。
   1. テキスト ファイルを作成し、名前を transactions_injector.py にします。
   2. transactions_injector.py に次のコードをコピーして貼り付けます。project-id は、実際のプロジェクト ID に置き換えます。

      #!/usr/bin/env python
      
      import datetime, json, os, random, time
      
      # Set the `project` variable to a Google Cloud project ID.
      project = 'project-id'
      
      FIRST_NAMES = ['Monet', 'Julia', 'Angelique', 'Stephane', 'Allan', 'Ulrike', 'Vella', 'Melia',
          'Noel', 'Terrence', 'Leigh', 'Rubin', 'Tanja', 'Shirlene', 'Deidre', 'Dorthy', 'Leighann',
          'Mamie', 'Gabriella', 'Tanika', 'Kennith', 'Merilyn', 'Tonda', 'Adolfo', 'Von', 'Agnus',
          'Kieth', 'Lisette', 'Hui', 'Lilliana',]
      CITIES = ['Washington', 'Springfield', 'Franklin', 'Greenville', 'Bristol', 'Fairview', 'Salem',
          'Madison', 'Georgetown', 'Arlington', 'Ashland',]
      STATES = ['MO','SC','IN','CA','IA','DE','ID','AK','NE','VA','PR','IL','ND','OK','VT','DC','CO','MS',
          'CT','ME','MN','NV','HI','MT','PA','SD','WA','NJ','NC','WV','AL','AR','FL','NM','KY','GA','MA',
          'KS','VI','MI','UT','AZ','WI','RI','NY','TN','OH','TX','AS','MD','OR','MP','LA','WY','GU','NH']
      PRODUCTS = ['Product 2', 'Product 2 XL', 'Product 3', 'Product 3 XL', 'Product 4', 'Product 4 XL', 'Product 5',
          'Product 5 XL',]
      
      while True:
        first_name, last_name = random.sample(FIRST_NAMES, 2)
        data = {
          'tr_time_str': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
          'first_name': first_name,
          'last_name': last_name,
          'city': random.choice(CITIES),
          'state':random.choice(STATES),
          'product': random.choice(PRODUCTS),
          'amount': float(random.randrange(50000, 70000)) / 100,
        }
      
        # For a more complete example on how to publish messages in Pub/Sub.
        #   https://cloud.google.com/pubsub/docs/publisher
        message = json.dumps(data)
        command = "gcloud --project={} pubsub topics publish transactions --message='{}'".format(project, message)
        print(command)
        os.system(command)
        time.sleep(random.randrange(1, 5))
      

BigQuery データセットとテーブルを作成する

1. BigQuery ウェブ UI で、BigQuery データセットを作成します。BigQuery データセットは、テーブルを格納する最上位のコンテナです。BigQuery のテーブルはデータセットに属している必要があります。
   1. 左側の [**エクスプローラ**] パネルで、プロジェクト ID の右側にあるアクション メニューを開きます。メニューで、[データセットを作成] をクリックします。次のスクリーンショットでは、dataflow-sql がプロジェクト ID です。

      

   2. 右側に表示された [データセットを作成] パネルで、データセット ID dataflow_sql_tutorial を入力します。[データセットを作成] をクリックします。

      

2. BigQuery テーブルを作成します。
   1. テキスト ファイルを作成し、名前を us_state_salesregions.csv にします。
   2. us_state_salesregions.csv に次のデータをコピーして貼り付けます。次のステップで、このデータを BigQuery テーブルに読み込みます。

      state_id,state_code,state_name,sales_region
      1,MO,Missouri,Region_1
      2,SC,South Carolina,Region_1
      3,IN,Indiana,Region_1
      6,DE,Delaware,Region_2
      15,VT,Vermont,Region_2
      16,DC,District of Columbia,Region_2
      19,CT,Connecticut,Region_2
      20,ME,Maine,Region_2
      35,PA,Pennsylvania,Region_2
      38,NJ,New Jersey,Region_2
      47,MA,Massachusetts,Region_2
      54,RI,Rhode Island,Region_2
      55,NY,New York,Region_2
      60,MD,Maryland,Region_2
      66,NH,New Hampshire,Region_2
      4,CA,California,Region_3
      8,AK,Alaska,Region_3
      37,WA,Washington,Region_3
      61,OR,Oregon,Region_3
      33,HI,Hawaii,Region_4
      59,AS,American Samoa,Region_4
      65,GU,Guam,Region_4
      5,IA,Iowa,Region_5
      32,NV,Nevada,Region_5
      11,PR,Puerto Rico,Region_6
      17,CO,Colorado,Region_6
      18,MS,Mississippi,Region_6
      41,AL,Alabama,Region_6
      42,AR,Arkansas,Region_6
      43,FL,Florida,Region_6
      44,NM,New Mexico,Region_6
      46,GA,Georgia,Region_6
      48,KS,Kansas,Region_6
      52,AZ,Arizona,Region_6
      56,TN,Tennessee,Region_6
      58,TX,Texas,Region_6
      63,LA,Louisiana,Region_6
      7,ID,Idaho,Region_7
      12,IL,Illinois,Region_7
      13,ND,North Dakota,Region_7
      31,MN,Minnesota,Region_7
      34,MT,Montana,Region_7
      36,SD,South Dakota,Region_7
      50,MI,Michigan,Region_7
      51,UT,Utah,Region_7
      64,WY,Wyoming,Region_7
      9,NE,Nebraska,Region_8
      10,VA,Virginia,Region_8
      14,OK,Oklahoma,Region_8
      39,NC,North Carolina,Region_8
      40,WV,West Virginia,Region_8
      45,KY,Kentucky,Region_8
      53,WI,Wisconsin,Region_8
      57,OH,Ohio,Region_8
      49,VI,United States Virgin Islands,Region_9
      62,MP,Commonwealth of the Northern Mariana Islands,Region_9
      

   3. BigQuery UI の左側にある [**エクスプローラ**] パネルで、プロジェクトの下に表示されている、作成済みの dataflow_sql_tutorial データセットのアクション メニューを開きます。[開く] をクリックします。
   4. 右側のウィンドウで [テーブルを作成] をクリックします。

      

   5. 右側に [テーブルの作成] パネルが表示されたら、次の操作を行います。
      1. [テーブルの作成元] プルダウン リストから [アップロード] を選択します。

         

      2. [ファイルを選択] では、[参照] をクリックし、作成した us_state_salesregions.csv ファイルを選択します。
      3. [テーブル名] ボックスに「us_state_salesregions」と入力します。
      4. [スキーマ] で、[自動検出] チェックボックスをオンにします。
      5. [詳細オプション] をクリックして、[詳細オプション] セクションを開きます。
      6. [スキップするヘッダー行] ボックスに「1」と入力します。

         

      7. [テーブルを作成] をクリックします。
   6. 左側のナビゲーション パネルで、プロジェクトの下にあるデータセットの [us_state_salesregions] をクリックします。[スキーマ] には、自動生成されたスキーマが表示されます。[プレビュー] には、テーブルのデータが表示されます。

      

Pub/Sub トピックにスキーマを割り当てる

スキーマを割り当てると、Pub/Sub トピックデータに SQL クエリを実行できます。現在、Dataflow SQL では、Pub/Sub トピックのメッセージは JSON 形式でシリアル化されます。

Pub/Sub トピックのサンプル transactions にスキーマを割り当てるには:

1. テキスト ファイルを作成し、名前を transactions_schema.yaml にします。次のスキーマ テキストをコピーして transactions_schema.yaml に貼り付けます。

  - column: event_timestamp
    description: Pub/Sub event timestamp
    mode: REQUIRED
    type: TIMESTAMP
  - column: tr_time_str
    description: Transaction time string
    mode: NULLABLE
    type: STRING
  - column: first_name
    description: First name
    mode: NULLABLE
    type: STRING
  - column: last_name
    description: Last name
    mode: NULLABLE
    type: STRING
  - column: city
    description: City
    mode: NULLABLE
    type: STRING
  - column: state
    description: State
    mode: NULLABLE
    type: STRING
  - column: product
    description: Product
    mode: NULLABLE
    type: STRING
  - column: amount
    description: Amount of transaction
    mode: NULLABLE
    type: FLOAT

1. gcloud コマンドライン ツールを使用してスキーマを割り当てます。

   a. 次のコマンドで gcloud ツールを更新します。gcloud ツールのバージョンは 242.0.0 以降にする必要があります。

     gcloud components update
   

   b. コマンドライン ウィンドウで次のコマンドを実行します。project-id はプロジェクト ID に置き換え、path-to-file は transactions_schema.yaml ファイルのパスに置き換えます。

     gcloud data-catalog entries update \
       --lookup-entry='pubsub.topic.`project-id`.transactions' \
       --schema-from-file=path-to-file/transactions_schema.yaml
   

   コマンドのパラメータと許可されるスキーマ ファイルの形式については、gcloud data-catalog entries update のドキュメント ページをご覧ください。

   注: ` を使用してプロジェクト ID の特殊文字をエスケープする必要があります。

   c. transactions Pub/Sub トピックにスキーマが正常に割り当てられたことを確認します。project-id は、実際のプロジェクト ID に置き換えます。

     gcloud data-catalog entries lookup 'pubsub.topic.`project-id`.transactions'
   

Pub/Sub ソースを検索する

Dataflow SQL UI では、アクセス権を持つプロジェクトの Pub/Sub データソース オブジェクトを検索できます。フルネームを覚える必要がありません。

このチュートリアルの例では、作成した transactions Pub/Sub トピックを検索します。

1. 左側のパネルで projectid=project-id transactions を検索します。project-id は、実際のプロジェクト ID に置き換えます。

   

スキーマを表示する

1. Dataflow SQL UI の左側のエクスプローラ パネルで、[トランザクション] をクリックするか、「projectid=project-id system=cloud_pubsub」と入力して Pub/Sub トピックを検索し、トピックを選択します。

2. [スキーマ] の下に、Pub/Sub トピックに割り当てられたスキーマが表示されます。

   

SQL クエリを作成する

Dataflow SQL UI で、Dataflow ジョブに実行する SQL クエリを作成します。

次の SQL クエリはデータ拡充クエリです。州と販売地域をマッピングする BigQuery テーブル（us_state_salesregions）を使用して、sales_region フィールドを Pub/Sub イベント ストリーム（transactions）に新たに追加します。

次の SQL クエリをコピーして、クエリエディタに貼り付けます。project-id は実際のプロジェクト ID に置き換えます。

SELECT tr.*, sr.sales_region
FROM pubsub.topic.`project-id`.transactions as tr
  INNER JOIN bigquery.table.`project-id`.dataflow_sql_tutorial.us_state_salesregions AS sr
  ON tr.state = sr.state_code

Dataflow SQL UI でクエリを入力すると、クエリ検証ツールによりクエリ構文が検証されます。クエリが有効な場合、緑色のチェックマーク アイコンが表示されます。クエリが無効な場合は、赤色の感嘆符アイコンが表示されます。クエリ構文が無効な場合は、検証ツールアイコンをクリックすると、修正が必要な箇所に関する情報が表示されます。

次のスクリーンショットでは、クエリエディタに有効なクエリが入力されています。緑色のチェックマークが表示されています。

SQL クエリを実行する Dataflow ジョブを作成する

サンプルソースを作成した場合: SQL クエリを実行する前に、Python スクリプトの transactions_injector.py を実行します。このスクリプトは、定期的にメッセージを Pub/Sub トピック transactions に公開します。スクリプトを停止するまで、トピックへのメッセージの公開を継続します。

python transactions_injector.py

SQL クエリを実行するには、Dataflow SQL UI から Dataflow ジョブを作成します。

1. クエリエディタの上にある [ジョブを作成] をクリックします。

2. 右側に表示された [Dataflow ジョブの作成] パネルの [送信先] オプションで、[BigQuery] を選択します。次に、[データセット ID] で dataflow_sql_tutorial を選択し、[テーブル名] を sales に設定します。

   

3. （省略可）Dataflow では、Dataflow SQL ジョブに最適な設定が自動的に選択されますが、オプション パラメータのメニューを展開して、次のパイプライン オプションを手動で指定することもできます。

   • ワーカーの最大数
   • ゾーン
   • サービス アカウントのメール
   • マシンタイプ
   • 追加テスト
   • ワーカー IP アドレスの構成
   • ネットワーク
   • サブネットワーク

4. [作成] をクリックします。数分後に Dataflow ジョブが開始します。

Dataflow ジョブを表示する

Dataflow は、SQL クエリを Apache Beam パイプラインに変換します。新しいブラウザタブに表示された Dataflow ウェブ UI で、パイプラインの詳細を確認できます。

ボックスをクリックすると、パイプラインで発生した変換の詳細が表示されます。たとえば、Run SQL Query というラベルのボックスをクリックすると、バックグラウンドで実行されたオペレーションが表示されます。

上の 2 つのボックスは、結合した 2 つの入力を表しています。1 つは Pub/Sub トピックの transactions で、もう 1 つは BigQuery テーブルの us_state_salesregions です。

ジョブ結果が含まれる出力テーブルを表示するには、BigQuery UI に移動してテーブルを表示します。左側のエクスプローラ パネルで、プロジェクトの下に表示されている、作成済みの dataflow_sql_tutorial データセットをクリックします。続いて、出力テーブルの sales をクリックします。出力テーブルの内容が [プレビュー] タブに表示されます。

以前に実行されたジョブを表示してクエリを編集する

Dataflow UI では、過去のジョブとクエリが [ジョブ] ページに保存されます。

ジョブ履歴リストを使用して、以前の SQL クエリを確認できます。たとえば、販売地域別の販売数を 15 秒ごとに集計するようにクエリを変更できます。[ジョブ] ページを使用して、このチュートリアルで開始した実行中のジョブにアクセスし、SQL クエリをコピーして、変更したクエリを使用して別のジョブを実行します。

1. Dataflow の [ジョブ] ページで、編集するジョブをクリックします。

2. [ジョブの詳細] ページの右側のパネルにある [パイプライン オプション] で SQL クエリを探します。queryString の行を見つけます。

   

3. SQL クエリをクエリエディタにコピーして貼り付け、タンブリング ウィンドウを追加します。次のクエリをコピーする場合は、project-id を実際のプロジェクト ID に置き換えます。

    SELECT
      sr.sales_region,
      TUMBLE_START("INTERVAL 15 SECOND") AS period_start,
      SUM(tr.amount) as amount
    FROM pubsub.topic.`project-id`.transactions AS tr
      INNER JOIN bigquery.table.`project-id`.dataflow_sql_tutorial.us_state_salesregions AS sr
      ON tr.state = sr.state_code
    GROUP BY
      sr.sales_region,
      TUMBLE(tr.event_timestamp, "INTERVAL 15 SECOND")
   

4. 変更したクエリを使用して新しいジョブを作成するには、[ジョブを作成] をクリックします。

クリーンアップ

このチュートリアルで使用したリソースについて、Cloud 請求先アカウントに課金されないようにする手順は次のとおりです。

1. transactions_injector.py 公開スクリプトが実行されている場合は、それを停止します。

2. 実行中の Dataflow ジョブを停止します。Cloud Console で Dataflow ウェブ UI に移動します。

   Dataflow ウェブ UI に移動

   このチュートリアルで作成したジョブごとに、次の操作を行います。

   1. ジョブの名前をクリックします。

   2. ジョブの [ジョブの概要] パネルで [ジョブを停止] をクリックします。[ジョブの停止] ダイアログが開き、ジョブの停止方法に関する選択肢が表示されます。

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

   4. [ジョブを停止] をクリックします。サービスはすべてのデータの取り込みと処理をできる限り早く停止します。[キャンセル] は処理を即時に停止するため、処理中のデータが失われる可能性があります。ジョブの停止には数分かかることがあります。

3. BigQuery データセットを削除します。Cloud Console の BigQuery ウェブ UI に移動します。

   BigQuery ウェブ UI に移動

   1. [エクスプローラ] パネルの [リソース] セクションで、作成した dataflow_sql_tutorial データセットをクリックします。

   2. 詳細パネルの右側で [データセットを削除] をクリックします。この操作を行うと、データセット、テーブル、すべてのデータが削除されます。

   3. [データセットの削除] ダイアログ ボックスでデータセットの名前（dataflow_sql_tutorial）を入力して、[削除] をクリックします。

4. Pub/Sub トピックを削除します。Cloud Console の Pub/Sub トピックページに移動します。

   Pub/Sub トピックページに移動

   1. transactions トピックの横にあるチェックボックスにチェックを入れます。

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

   3. Pub/Sub サブスクリプション ページに移動します。

   4. transactions の残っているサブスクリプションの横にあるチェックボックスにチェックを入れます。実行中のジョブがない場合、サブスクリプションが表示されない可能性があります。

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

5. Cloud Storage で Dataflow ステージング バケットを削除します。Cloud Console の Cloud Storage ブラウザに移動します。

   Cloud Storage ブラウザに移動

   1. Dataflow ステージング バケットの隣にあるチェックボックスをオンにします。

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

次のステップ

• Dataflow SQL の概要を確認する。
• ストリーミング パイプラインの基本について学ぶ。
• データのソースと宛先の使用について読む。
• Dataflow SQL リファレンスを確認する。
• Cloud Next 2019 で行われたストリーミング分析のデモを見る。