サービスを選ぶ
データ処理と分析
データの流れを、取り込み、加工、分析、順序の管理の 4 段に分けて、段ごとにサービスを選びます。イベントの取り込みは Pub/Sub、新しく書く加工は Dataflow、既存の Spark や Hadoop の処理は Managed Service for Apache Spark (旧 Dataproc)、分析は BigQuery が基本です。
データ処理の問題は、データの出どころから分析までを 1 本の流れとして描けるかを問います。流れを取り込み、加工、分析、順序の管理の 4 段に分け、段ごとにサービスを 1 つ当てはめます。
バッチかストリーミングか
最初に、データが途切れなく届くストリーミングか、まとまって届くバッチかを決めます。問題文に「リアルタイム」「数秒以内に検知」とあればストリーミング、「毎晩」「日次」「既存のバッチジョブ」とあればバッチです。Dataflow は同じコードでどちらも扱えるので、両方が混ざる要件でも選べます。
| 段 | 新しく作るとき | 既存の資産を活かすとき |
|---|---|---|
| 取り込み (イベント) | Pub/Sub | Pub/Sub (Kafka などからの置き換え) |
| 取り込み (ファイル) | Storage Transfer Service、Transfer Appliance | 同左 |
| 加工 | Dataflow | Managed Service for Apache Spark |
| 分析 | BigQuery | BigQuery |
| 順序と依存の管理 | Managed Service for Apache Airflow | 同左 (既存の DAG をそのまま使える) |
加工で迷ったら、既存の Spark や Hadoop のジョブがあるかで決めます。なければ Dataflow です。ゼロから作るパイプラインに Spark のクラスタを立てる構成は選びません。
Pub/Sub: サブスクリプションごとに全件届く
Pub/Sub では、送り手 (パブリッシャー) がトピックにメッセージを送り、受け手 (サブスクライバー) はトピックに付けたサブスクリプションから受け取ります。受け手が止まっても負荷が急増しても、送り手には影響しません。
サブスクリプションの設定では、確認応答の期限は既定 10 秒、最長 600 秒です。期限を過ぎたメッセージはもう一度配信されるので、受け手は同じメッセージが 2 回届いても困らないように作ります。確認応答されなかったメッセージは保持期間 (既定 7 日、最長 31 日) のあいだ残ります。順序キーを付けて順序指定を有効にすると、同じリージョンで送られた同じキーのメッセージは送った順に届きます。
加工せずに BigQuery の表へ入れるだけなら、BigQuery サブスクリプションで Dataflow なしに直接書き込めます。時間の窓ごとに集計するなら Dataflow を挟みます。
加工は Dataflow か Managed Service for Apache Spark
Dataflow は Apache Beam で書いたパイプラインを実行するマネージドなサービスで、ワーカーの台数を処理量に合わせて自動で増減します。Beam はバッチとストリーミングを同じモデルで書けます。5 分ごとの売上のような窓の集計と、遅れて届いたデータの扱いも Beam の機能にあり、Pub/Sub から BigQuery へ流す典型的な処理は Google のテンプレートから始められます。
Managed Service for Apache Spark は、Spark、Hadoop、Hive などをマネージドで動かすサービスです。公式ドキュメントでは、以前の Dataproc (クラスタの形) と Serverless for Apache Spark (サーバーレスの形) がこの名前にまとめられています。古い教材の Dataproc は同じサービスです。オンプレミスの Spark のジョブを、コードをほぼ変えずに移すときに選びます。ジョブのときだけクラスタを作って終われば消し、データは HDFS ではなく Cloud Storage に置くと、クラスタを消してもデータが残ります。
分析は BigQuery、順序は Airflow
BigQuery は SQL で大量のデータを分析するデータウェアハウスです。先に BigQuery に入れて SQL で加工する ELT も、Cloud Storage のファイルを外部テーブルとして直接クエリすることもできます。行単位で頻繁に更新するアプリの DB には使いません (ストレージとデータベースの選び方)。機械学習のモデル作りはAI と機械学習のサービスで扱います。
Managed Service for Apache Airflow は Apache Airflow のマネージドサービスで、Cloud Composer がこの名前へ移行していると告知されています。Python で書いた DAG で、「ファイルの到着を待つ、Spark を動かす、BigQuery で集計する、失敗したら通知する」のような複数サービスにまたがる順序と依存を管理します。Airflow はデータを加工せず、ほかのサービスに順番に仕事をさせる道具です。処理が Pub/Sub から BigQuery への 1 本だけなら要りません。
大量のデータを移す手段は日数で決める
Storage Transfer Service は、Amazon S3、Azure Blob Storage、オンプレミスのファイルシステム、HDFS などから Cloud Storage へネットワーク経由で移します。オンプレミスからは手元にエージェントを置き、他のクラウドからはエージェントなしで転送します。Transfer Appliance は、Google から届く機器にデータを書き込んで送り返す方法です。
1 週間以内に終わるなら Storage Transfer Service、終わらないなら Transfer Appliance です。回線を太くする選択肢はオンプレミス・マルチクラウドとの接続で扱います。
試してみる
Cloud Shell で Pub/Sub のメッセージを送って受け取り、BigQuery の一般公開データセットに SQL を投げます。無料枠の範囲で終わります。
gcloud services enable pubsub.googleapis.com bigquery.googleapis.com
gcloud pubsub topics create orders
gcloud pubsub subscriptions create orders-sub --topic=orders
gcloud pubsub topics publish orders --message='{"order_id": 1, "amount": 1200}'
gcloud pubsub topics publish orders --message='{"order_id": 2, "amount": 800}'
gcloud pubsub subscriptions pull orders-sub --auto-ack --limit=10もう一度 pull すると、確認済みのメッセージは返ってきません。
bq query --use_legacy_sql=false \
'SELECT name, SUM(number) AS total
FROM `bigquery-public-data.usa_names.usa_1910_2013`
GROUP BY name ORDER BY total DESC LIMIT 10'終わったらサブスクリプションとトピックを削除します。
gcloud pubsub subscriptions delete orders-sub
gcloud pubsub topics delete orders試験で問われる判断
以下は、このカリキュラム用に作った例です。
「注文イベントを受け、5 分ごとの売上をダッシュボードに出す。セールの日の急増も取りこぼしたくない」。答えは Pub/Sub で受け、Dataflow で 5 分の窓ごとに集計し、BigQuery に書く構成です。
「オンプレミスの Hadoop で数百本の Spark のジョブが動いている。改修を最小にして移したい」。答えは Managed Service for Apache Spark で、データは Cloud Storage に移します。Dataflow は Beam への書き直しが要ります。
「500 TB をオンプレミスから移す。回線は 1 Gbps で、業務にも使っている」。答えは Transfer Appliance です。回線を使い切っても 40 日以上かかります。
「毎晩 3 つのシステムからファイルが届くのを待ち、Spark で結合し、BigQuery に読み込み、失敗したら通知する」。答えは Managed Service for Apache Airflow の DAG です。
次のページでは、ここまでのサービスをつなぐネットワークを VPC の設計で扱います。