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

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

目標

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

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

費用

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

Dataflow
Cloud Storage
Pub/Sub

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

始める前に

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

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

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

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

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

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

サービスアカウントを作成します。
1. Cloud Console で [サービスアカウントの作成] ページに移動します。
  [サービスアカウントの作成] に移動
2. プロジェクトを選択します。
3. [サービスアカウント名] フィールドに名前を入力します。Cloud Console は、この名前に基づいて [サービスアカウント ID] フィールドに入力します。
  
  [サービスアカウントの説明] フィールドに説明を入力します。例: Service account for quickstart
4. [作成] をクリックします。
5. [ロールを選択] フィールドをクリックします。
  
  [クイックアクセス] で [基本]、[オーナー] の順にクリックします。
  
  注: [ロール] フィールドは、サービスアカウントがプロジェクト内のどのリソースにアクセスできるかに影響します。これらのロールは後で取り消すことも、追加のロールを付与することもできます。本番環境では、オーナー、編集者、閲覧者のロールを付与しないでください。詳細については、リソースへのアクセス権の付与、変更、取り消しをご覧ください。
6. [続行] をクリックします。
7. [完了] をクリックして、サービスアカウントの作成を完了します。
  
  ブラウザウィンドウは閉じないでください。次のステップでこれを使用します。
サービスアカウントキーを作成します。
1. Cloud Console で、作成したサービスアカウントのメールアドレスをクリックします。
2. [キー] をクリックします。
3. [鍵を追加]、[新しい鍵を作成] の順にクリックします。
4. [作成] をクリックします。JSON キーファイルがパソコンにダウンロードされます。
5. [閉じる] をクリックします。

環境変数 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 ファイルのパスに置き換えます。

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

Dataflow SQL UI に切り替える

BigQuery ウェブ UI で、次の手順に沿って Dataflow UI に切り替えます。

[展開] プルダウンメニューを開いて、[クエリの設定] を選択します。
右側に表示された [クエリの設定] メニューで、[Dataflow エンジン] を選択します。
プロジェクトで Dataflow と Data Catalog API が有効になっていない場合、これらを有効にするように求められます。[API を有効にする] をクリックします。Dataflow API と Data Catalog API の有効化には数分かかる場合があります。
API が有効になったら [保存] をクリックします。

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

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

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

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

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

Pub/Sub トピックにメッセージを公開する Python スクリプトを作成します。まだインストールしていない場合は Python をインストールする必要があります。このスクリプトは、SQL クエリを実行する直前にコマンドラインウィンドウで実行します。

テキストファイルを作成し、名前を transactions_injector.py にします。

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 データセットとテーブルを作成する

BigQuery ウェブ UI で、BigQuery データセットを作成します。BigQuery データセットは、テーブルを格納する最上位のコンテナです。BigQuery のテーブルはデータセットに属している必要があります。
1. 左側のナビゲーションパネルで、プロジェクト ID をクリックします。右側のウィンドウで [データセットを作成] をクリックします。次のスクリーンショットでは、dataflow-sql がプロジェクト ID です。
2. 右側に表示された [データセットを作成] パネルで、データセット ID dataflow_sql_datasetを入力します。[データセットを作成] をクリックします。
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. Dataflow SQL UI の左側のナビゲーションパネルで、プロジェクトの下に作成した dataflow_sql_dataset データセットをクリックします。
4. 右側のウィンドウで [テーブルを作成] をクリックします。
5. 右側に [テーブルの作成] パネルが表示されたら、次の操作を行います。
  1. [テーブルの作成元] プルダウンリストから [アップロード] を選択します。
  2. [ファイルを選択] では、[参照] をクリックし、作成した us_state_salesregions.csv ファイルを選択します。
  3. [テーブル名] ボックスに「us_state_salesregions」と入力します。
  4. [スキーマ] で、[自動検出] チェックボックスをオンにします。
  5. [詳細オプション] をクリックして、[詳細オプション] セクションを展開します。
  6. [スキップするヘッダー行] ボックスに「1」と入力します。
  7. [テーブルを作成] をクリックします。
6. 左側のナビゲーションパネルで、プロジェクトの下のデータセットにある us_state_salesregions をクリックします。[スキーマ] には、自動生成されたスキーマが表示されます。[プレビュー] には、テーブルのデータが表示されます。

Pub/Sub ソースを検索する

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

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

左側のナビゲーションパネルで、[データを追加] プルダウンリストをクリックして [Cloud Dataflow のソース] を選択します。
右側に表示される [Cloud Dataflow ソースを追加] パネルで、[Pub/Sub トピック] を選択します。検索ボックスで、「transactions」を検索します。トピックを選択して [追加] をクリックします。

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

スキーマを割り当てると、Pub/Sub トピックデータに SQL クエリを実行できます。現在、Dataflow SQL では、Pub/Sub トピックのメッセージは JSON 形式でシリアル化されます。今後、他の形式（Avro など）のサポートが追加される予定です。

サンプルの Pub/Sub トピックを Dataflow ソースとして追加した後、次の手順で Dataflow SQL UI のトピックにスキーマを割り当てます。

[リソース] パネルでトピックを選択します。
[スキーマ] タブで [スキーマを編集] をクリックします。右側に [スキーマ] サイドパネルが開きます。

[テキストとして編集] ボタンを切り替え、次のインラインスキーマをエディタに貼り付けます。[送信] をクリックします。

[
  {
      "description": "Pub/Sub event timestamp",
      "name": "event_timestamp",
      "mode": "REQUIRED",
      "type": "TIMESTAMP"
  },
  {
      "description": "Transaction time string",
      "name": "tr_time_str",
      "type": "STRING"
  },
  {
      "description": "First name",
      "name": "first_name",
      "type": "STRING"
  },
  {
      "description": "Last name",
      "name": "last_name",
      "type": "STRING"
  },
  {
      "description": "City",
      "name": "city",
      "type": "STRING"
  },
  {
      "description": "State",
      "name": "state",
      "type": "STRING"
  },
  {
      "description": "Product",
      "name": "product",
      "type": "STRING"
  },
  {
      "description": "Amount of transaction",
      "name": "amount",
      "type": "FLOAT64"
  }
]

（省略可）[トピックをプレビュー] をクリックしてメッセージの内容を確認し、定義したスキーマと一致していることを確認します。

スキーマを表示する

Dataflow SQL UI の左側のナビゲーションパネルで、[Cloud Dataflow のソース] をクリックします。
[Pub/Sub トピック] をクリックします。
[トランザクション] をクリックします。
[スキーマ] の下に、transactions 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_dataset.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 ジョブを作成します。

クエリエディタで [Dataflow ジョブを作成] をクリックします。
右側に表示された [Dataflow ジョブを作成] パネルで、デフォルトのテーブル名を dfsqltable_sales に変更します。
（省略可）Dataflow では、Dataflow SQL ジョブに最適な設定が自動的に選択されますが、オプションパラメータのメニューを展開して、次のパイプラインオプションを手動で指定することもできます。
- ワーカーの最大数
- ゾーン
- サービスアカウントのメール
- マシンタイプ
- 追加テスト
- ワーカー IP アドレスの構成
- ネットワーク
- サブネットワーク
[作成] をクリックします。数分後に Dataflow ジョブが開始します。
UI に [クエリ結果] パネルが表示されます。後でジョブの [クエリ結果] パネルに戻るには、[ジョブの履歴] パネルでジョブを検索し、[クエリをエディタで開く] ボタンを使用します。詳しくは、Dataflow ジョブと出力を表示するをご覧ください。
[ジョブ情報] でジョブ ID のリンクをクリックします。これで、Dataflow ウェブ UI で新しいブラウザタブが開き、Dataflow ジョブの詳細ページが表示されます。

Dataflow ジョブと出力を表示する

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

Dataflow ウェブ UI に表示された SQL クエリからのパイプライン。

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

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

ジョブ結果が含まれる出力テーブルを表示するには、Dataflow SQL UI のブラウザタブに戻ります。左側のナビゲーションパネルで、プロジェクトの下に表示されている、作成した dataflow_sql_dataset データセットをクリックします。続いて、出力テーブルの dfsqltable_sales をクリックします。[プレビュー] タブに出力テーブルの内容が表示されます。

dfsqltable_sales のプレビューテーブルには、tr_time_str、first_name、last_name、city、state、product、amount、sales_region の列があります。

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

Dataflow SQL UI では、以前に実行されたジョブとクエリを [ジョブの履歴] パネルで確認できます。ジョブは、ジョブを開始した日付順に表示されます。ジョブリストでは、実行中のジョブがある日付が先に表示され、その後に、実行中のジョブがない日付が続きます。

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

左側のナビゲーションパネルで、[ジョブの履歴] をクリックします。
[ジョブの履歴] で Cloud Dataflow をクリックします。プロジェクトで実行した以前のジョブがすべて表示されます。
編集するジョブをクリックします。[クエリエディタで開く] をクリックします。

[クエリエディタ] で 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_dataset.us_state_salesregions AS sr
   ON tr.state = sr.state_code
 GROUP BY
   sr.sales_region,
   TUMBLE(tr.event_timestamp, "INTERVAL 15 SECOND")

クエリエディタで [Cloud Dataflow ジョブを作成] をクリックし、変更したクエリを使用して新しいジョブを作成します。

クリーンアップ

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

transactions_injector.py 公開スクリプトが実行されている場合は、それを停止します。
実行中の Dataflow ジョブを停止します。Cloud Console で Dataflow ウェブ UI に移動します。

Dataflow ウェブ UI に移動

このチュートリアルで作成したジョブごとに、次の操作を行います。
1. ジョブの名前をクリックします。
2. ジョブの [ジョブの概要] パネルで [ジョブを停止] をクリックします。[ジョブの停止] ダイアログが開き、ジョブの停止方法に関する選択肢が表示されます。
3. [キャンセル] をクリックします。
4. [ジョブを停止] をクリックします。サービスはすべてのデータの取り込みと処理をできる限り早く停止します。[キャンセル] は処理を即時に停止するため、処理中のデータが失われる可能性があります。ジョブの停止には数分かかることがあります。
BigQuery データセットを削除します。Cloud Console の BigQuery ウェブ UI に移動します。

BigQuery ウェブ UI に移動
1. ナビゲーションパネルの [リソース] セクションで、作成した dataflow_sql_dataset データセットをクリックします。
2. 詳細パネルの右側で [データセットを削除] をクリックします。この操作を行うと、データセット、テーブル、すべてのデータが削除されます。
3. [データセットの削除] ダイアログボックスでデータセットの名前（dataflow_sql_dataset）を入力して、[削除] をクリックします。
Pub/Sub トピックを削除します。Cloud Console の Pub/Sub トピックページに移動します。

Pub/Sub トピックページに移動
1. transactions トピックの横にあるチェックボックスにチェックを入れます。
2. [削除] をクリックして、トピックを永続的に削除します。
3. Pub/Sub サブスクリプションページに移動します。
4. transactions の残っているサブスクリプションの横にあるチェックボックスにチェックを入れます。実行中のジョブがない場合、サブスクリプションが表示されない可能性があります。
5. [削除] をクリックして、サブスクリプションを永続的に削除します。
Cloud Storage で Dataflow ステージングバケットを削除します。Cloud Console の Cloud Storage ブラウザに移動します。

Cloud Storage ブラウザに移動
1. Dataflow ステージングバケットの隣にあるチェックボックスをオンにします。
2. [削除] をクリックして、バケットを永続的に削除します。

次のステップ

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