このページでは、Google 提供の Cloud Dataflow テンプレートを使用してストリーミング パイプラインを作成する方法を説明します。具体的には、このページでは例として Cloud Pub/Sub Topic to BigQuery テンプレートを使用します。
始める前に
-
Google アカウントにログインします。
Google アカウントをまだお持ちでない場合は、新しいアカウントを登録します。
-
GCP Console のプロジェクト セレクタのページで、GCP プロジェクトを選択または作成します。
-
Google Cloud Platform プロジェクトに対して課金が有効になっていることを確認します。 詳しくは、課金を有効にする方法をご覧ください。
- Cloud Dataflow、Compute Engine、Stackdriver Logging、Google Cloud Storage、Google Cloud Storage JSON、BigQuery、Cloud Pub/Sub、Cloud Resource Manager API を有効にします。
- Cloud Storage バケットを作成します。
- GCP Console で、Cloud Storage ブラウザページに移動します。
- [バケットを作成] をクリックします。
- [バケットを作成] ダイアログ内で、以下の属性を指定します。
- 名前: 一意のバケット名。バケットの名前空間は、全世界で、誰もが見られるようになっていますので、機密情報をバケット名に含めないようにしてください。
- デフォルト ストレージ クラス: Standard
- バケットデータが保存されるロケーション。
- [作成] をクリックします。
Cloud BigQuery データセットとテーブルの作成
自身の Cloud Pub/Sub トピックに適したスキーマを使用して、Google Cloud Shell または GCP Console で BigQuery データセットとテーブルを作成します。
この例で、データセットの名前は taxirides、テーブルの名前は realtime です。
Cloud Shell の使用
Cloud Shell を使用してデータセットとテーブルを作成します。
- 次のコマンドを実行して、データセットを作成します。
bq mk taxirides
出力は次のようになります。Dataset “myprojectid:taxirides” successfully created
- 次のコマンドを実行して、テーブルを作成します。
bq mk \ --time_partitioning_field timestamp \ --schema ride_id:string,point_idx:integer,latitude:float,longitude:float,\ timestamp:timestamp,meter_reading:float,meter_increment:float,ride_status:string,\ passenger_count:integer -t taxirides.realtime
出力は次のようになります。Table “myprojectid:taxirides.realtime” successfully created
テーブルは、クエリ費用を削減してパフォーマンスを向上させるために分割されます。
Google Cloud Platform Console の使用
Google Cloud Platform Console を使用して、データセットとテーブルを作成します。
- BigQuery ウェブ UI に移動します。
BigQuery ウェブ UI に移動 - ナビゲーションでプロジェクト名の横にある下矢印アイコンをクリックし、[Create new dataset] をクリックします。データセット ID として「
taxirides」を入力します。
データセット ID は、プロジェクトごとに一意です。疑問符アイコンをクリックすると、ID の制限について確認できます。
- その他のデフォルト設定はすべてそのままにし、[OK] をクリックします。
- ナビゲーションで、作成したデータセット ID の上にカーソルを合わせます。ID の横にある下矢印アイコンをクリックし、[Create new table] をクリックします。
- [Source Data] の横にある [Create empty table] オプションを選択します。
- [Destination Table] で
taxiridesを選択し、「realtime」と入力します。 - [Schema] で [Edit as Text] を選択し、次のように入力します。
ride_id:string,point_idx:integer,latitude:float,longitude:float,timestamp:timestamp, meter_reading:float,meter_increment:float,ride_status:string,passenger_count:integer
- [Options] の [Partitioning Type] フィールドで [Day] オプションを選択します。
- [Options] の [Partitioning Field] セレクタで [timestamp] 列を選択します。
- [テーブルを作成] ボタンをクリックします。
パイプラインを実行する
Google が提供する Cloud Pub/Sub Topic to BigQuery テンプレートを使用して、ストリーミング パイプラインを実行します。
- Cloud Dataflow ウェブ UI に移動します。
Cloud Dataflow ウェブ UI に移動 - [テンプレートからジョブを作成] をクリックします。
- Cloud Dataflow ジョブの [ジョブ名] を入力します。
- [Cloud Dataflow テンプレート] で、[Cloud Pub/Sub Topic to BigQuery] テンプレートを選択します。
- [Cloud Pub/Sub 入力トピック] に、「
projects/pubsub-public-data/topics/taxirides-realtime」を入力します。パイプラインは入力トピックから受信データを取得します。 - [BigQuery 出力テーブル] に「
<myprojectid>:taxirides.realtime」を入力します。 - [一時的なロケーション] に「
gs://<mybucket>/tmp/」を入力します。これは、ステージング済みパイプライン ジョブと同様に、一時ファイルを保存するためのサブフォルダです。 - [ジョブを実行] ボタンをクリックします。
- BigQuery に書き込まれたデータを表示します。BigQuery ウェブ UI に移動します。
BigQuery ウェブ UI に移動
クエリは、標準 SQL を使用して送信できます。たとえば次のクエリでは、過去 24 時間以内に追加されたすべての行が選択されます。SELECT * FROM `myprojectid.taxirides.realtime` WHERE `timestamp` > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY) LIMIT 1000
クリーンアップ
このチュートリアルで使用したリソースについて GCP アカウントに課金されないようにするために、以下を行ってください。
- Cloud Dataflow ウェブ UI に移動します。
Cloud Dataflow ウェブ UI に移動 - BigQuery ウェブ UI に移動します。
BigQuery ウェブ UI に移動- ナビゲーションで、作成した taxirides データセットの上にカーソルを合わせます。
- ナビゲーションのデータセット名の横にある下矢印アイコンをクリックし、[データセットの削除] をクリックします。
- [データセットの削除] ダイアログ ボックスでデータセットの名前（taxirides）を入力し、[OK] をクリックして確定します。
ご不明な点がありましたら、Google の