クイックスタート: Dataflow によるストリーム処理

Dataflow は、ストリーミング（リアルタイム）モードのデータとバッチモードのデータを同等の信頼性と表現力で変換、活用するフルマネージド サービスです。Apache Beam SDK を使用して、簡素化されたパイプライン開発環境を提供します。Apache Beam SDK は、ウィンドウ処理とセッション分析のプリミティブが豊富に用意されているだけでなく、ソースとシンクのコネクタからなるエコシステムも提供しています。このクイックスタートでは、Dataflow を使用して次の操作を行う方法を説明します。

  • Pub/Sub トピックにパブリッシュされたメッセージを読む
  • タイムスタンプでメッセージをウィンドウ処理（グループ化）する
  • Cloud Storage にメッセージを書き込む

このクイックスタートでは、Java と Python で Dataflow を使用する方法について説明します。SQL もサポートされます。

カスタムデータ処理を行う予定がない場合は、UI ベースの Dataflow テンプレートを使用することもできます。

準備

  1. Cloud SDK のインストールと初期化の手順に従って操作します。
  2. プロジェクトの課金を有効にします。

  3. このクイックスタートを完了するには、次の API を有効にする必要があります。Compute Engine、Google Cloud のオペレーション スイート、Cloud Storage、Cloud Storage JSON、Pub/Sub、Cloud Scheduler、Resource Manager、App Engine です。

    API を有効にする

    API がコンソールに表示されるまで数分かかることがあります。

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

    サービス アカウント キーを作成する

    1. [サービス アカウント] リストから [新しいサービス アカウント] を選択します。
    2. [サービス アカウント名] フィールドに名前を入力します。
    3. [役割] リストで、[プロジェクト] > [オーナー] を選択します。
    4. [作成] をクリックします。

    キーがブラウザのデフォルトのダウンロード フォルダに送信されます。

  5. サービス アカウント キーを指すように GOOGLE_APPLICATION_CREDENTIALS 環境変数を設定します。

    export GOOGLE_APPLICATION_CREDENTIALS=path/to/my/credentials.json

  6. バケットとプロジェクトの変数を作成します。Cloud Storage バケット名は、グローバルに一意である必要があります。

    BUCKET_NAME=bucket-name
PROJECT_NAME=$(gcloud config get-value project)

  7. このプロジェクトが所有する Cloud Storage バケットを作成します。

    gsutil mb gs://$BUCKET_NAME

  8. このプロジェクトで Pub/Sub トピックを作成します。

    gcloud pubsub topics create cron-topic

  9. このプロジェクトで Cloud Scheduler ジョブを作成します。このジョブは、1 分間隔で Cloud Pub/Sub トピックにメッセージをパブリッシュします。

    App Engine アプリがプロジェクトに存在しない場合、この手順で作成されます。

    gcloud scheduler jobs create pubsub publisher-job --schedule="* * * * *" \
 --topic=cron-topic --message-body="Hello!"

    ジョブを開始します。

    gcloud scheduler jobs run publisher-job

  10. 次のコマンドを使用して、クイックスタート リポジトリのクローンを作成し、サンプルコード ディレクトリに移動します。

    Java

    git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/pubsub/streaming-analytics

    Python

    git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
cd python-docs-samples/pubsub/streaming-analytics
pip install -r requirements.txt  # Install Apache Beam dependencies

Pub/Sub から Cloud Storage へのメッセージのストリーミング

コードサンプル

このサンプルコードでは、Dataflow を使用して次のことを行います。

  • Pub/Sub メッセージを読み取ります。
  • パブリッシュ タイムスタンプにより、固定サイズの間隔でメッセージをウィンドウ処理（グループ化）します。

  • 各ウィンドウのメッセージを Cloud Storage のファイルに書き込みます。

Java

import org.apache.beam.examples.common.WriteOneFilePerWindow;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

import java.io.IOException;

public class PubSubToGCS {
  /*
  * Define your own configuration options. Add your own arguments to be processed
  * by the command-line parser, and specify default values for them.
  */
  public interface PubSubToGCSOptions extends PipelineOptions, StreamingOptions {
    @Description("The Cloud Pub/Sub topic to read from.")
    @Required
    String getInputTopic();
    void setInputTopic(String value);

    @Description("Output file's window size in number of minutes.")
    @Default.Integer(1)
    Integer getWindowSize();
    void setWindowSize(Integer value);

    @Description("Path of the output file including its filename prefix.")
    @Required
    String getOutput();
    void setOutput(String value);
  }

  public static void main(String[] args) throws IOException {
    // The maximum number of shards when writing output.
    int numShards = 1;

    PubSubToGCSOptions options = PipelineOptionsFactory
      .fromArgs(args)
      .withValidation()
      .as(PubSubToGCSOptions.class);

    options.setStreaming(true);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
      // 1) Read string messages from a Pub/Sub topic.
      .apply("Read PubSub Messages", PubsubIO.readStrings().fromTopic(options.getInputTopic()))
      // 2) Group the messages into fixed-sized minute intervals.
      .apply(Window.into(FixedWindows.of(Duration.standardMinutes(options.getWindowSize()))))
      // 3) Write one file to GCS for every window of messages.
      .apply("Write Files to GCS", new WriteOneFilePerWindow(options.getOutput(), numShards));

    // Execute the pipeline and wait until it finishes running.
    pipeline.run().waitUntilFinish();
  }
}

Python

import argparse
import datetime
import json
import logging

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
import apache_beam.transforms.window as window

class GroupWindowsIntoBatches(beam.PTransform):
    """A composite transform that groups Pub/Sub messages based on publish
    time and outputs a list of dictionaries, where each contains one message
    and its publish timestamp.
    """

    def __init__(self, window_size):
        # Convert minutes into seconds.
        self.window_size = int(window_size * 60)

    def expand(self, pcoll):
        return (
            pcoll
            # Assigns window info to each Pub/Sub message based on its
            # publish timestamp.
            | "Window into Fixed Intervals"
            >> beam.WindowInto(window.FixedWindows(self.window_size))
            | "Add timestamps to messages" >> beam.ParDo(AddTimestamps())
            # Use a dummy key to group the elements in the same window.
            # Note that all the elements in one window must fit into memory
            # for this. If the windowed elements do not fit into memory,
            # please consider using `beam.util.BatchElements`.
            # https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.util.html#apache_beam.transforms.util.BatchElements
            | "Add Dummy Key" >> beam.Map(lambda elem: (None, elem))
            | "Groupby" >> beam.GroupByKey()
            | "Abandon Dummy Key" >> beam.MapTuple(lambda _, val: val)
        )

class AddTimestamps(beam.DoFn):
    def process(self, element, publish_time=beam.DoFn.TimestampParam):
        """Processes each incoming windowed element by extracting the Pub/Sub
        message and its publish timestamp into a dictionary. `publish_time`
        defaults to the publish timestamp returned by the Pub/Sub server. It
        is bound to each element by Beam at runtime.
        """

        yield {
            "message_body": element.decode("utf-8"),
            "publish_time": datetime.datetime.utcfromtimestamp(
                float(publish_time)
            ).strftime("%Y-%m-%d %H:%M:%S.%f"),
        }

class WriteBatchesToGCS(beam.DoFn):
    def __init__(self, output_path):
        self.output_path = output_path

    def process(self, batch, window=beam.DoFn.WindowParam):
        """Write one batch per file to a Google Cloud Storage bucket. """

        ts_format = "%H:%M"
        window_start = window.start.to_utc_datetime().strftime(ts_format)
        window_end = window.end.to_utc_datetime().strftime(ts_format)
        filename = "-".join([self.output_path, window_start, window_end])

        with beam.io.gcp.gcsio.GcsIO().open(filename=filename, mode="w") as f:
            for element in batch:
                f.write("{}\n".format(json.dumps(element)).encode("utf-8"))

def run(input_topic, output_path, window_size=1.0, pipeline_args=None):
    # `save_main_session` is set to true because some DoFn's rely on
    # globally imported modules.
    pipeline_options = PipelineOptions(
        pipeline_args, streaming=True, save_main_session=True
    )

    with beam.Pipeline(options=pipeline_options) as pipeline:
        (
            pipeline
            | "Read PubSub Messages"
            >> beam.io.ReadFromPubSub(topic=input_topic)
            | "Window into" >> GroupWindowsIntoBatches(window_size)
            | "Write to GCS" >> beam.ParDo(WriteBatchesToGCS(output_path))
        )

if __name__ == "__main__":  # noqa
    logging.getLogger().setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input_topic",
        help="The Cloud Pub/Sub topic to read from.\n"
        '"projects/<PROJECT_NAME>/topics/<TOPIC_NAME>".',
    )
    parser.add_argument(
        "--window_size",
        type=float,
        default=1.0,
        help="Output file's window size in number of minutes.",
    )
    parser.add_argument(
        "--output_path",
        help="GCS Path of the output file including filename prefix.",
    )
    known_args, pipeline_args = parser.parse_known_args()

    run(
        known_args.input_topic,
        known_args.output_path,
        known_args.window_size,
        pipeline_args,
    )

パイプラインの開始

パイプラインを開始するには、次のコマンドを使用します。

Java

mvn compile exec:java \
  -Dexec.mainClass=com.examples.pubsub.streaming.PubSubToGCS \
  -Dexec.cleanupDaemonThreads=false \
  -Dexec.args=" \
    --project=$PROJECT_NAME \
    --inputTopic=projects/$PROJECT_NAME/topics/cron-topic \
    --output=gs://$BUCKET_NAME/samples/output \
    --runner=DataflowRunner \
    --windowSize=2"

Python

python PubSubToGCS.py \
  --project=$PROJECT_NAME \
  --input_topic=projects/$PROJECT_NAME/topics/cron-topic \
  --output_path=gs://$BUCKET_NAME/samples/output \
  --runner=DataflowRunner \
  --window_size=2 \
  --temp_location=gs://$BUCKET_NAME/temp

ジョブとパイプラインの進行状況の確認

ジョブの進行状況は Dataflow コンソールで確認できます。

Dataflow コンソールに移動

ジョブの進行状況の確認

[ジョブの詳細] ビューを開いて、次の情報を確認します。

  • ジョブの構成
  • ジョブのログ
  • ステージ指標

ジョブの進行状況の確認

Cloud Storage に出力ファイルが表示されるまで数分間かかる場合があります。

ジョブの進行状況の確認

または、以下のコマンドラインを使用して、書き出されたファイルを確認します。

gsutil ls gs://${BUCKET_NAME}/samples/

出力は次のようになります。

Java

gs://{$BUCKET_NAME}/samples/output-22:30-22:32-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:32-22:34-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:34-22:36-0-of-1
gs://{$BUCKET_NAME}/samples/output-22:36-22:38-0-of-1

Python

gs://{$BUCKET_NAME}/samples/output-22:30-22:32
gs://{$BUCKET_NAME}/samples/output-22:32-22:34
gs://{$BUCKET_NAME}/samples/output-22:34-22:36
gs://{$BUCKET_NAME}/samples/output-22:36-22:38

クリーンアップ

  1. Cloud Scheduler ジョブを削除します。

    gcloud scheduler jobs delete publisher-job

  2. Ctrl+C を使用して、ターミナルでプログラムを停止します。

  3. Dataflow コンソールで、ジョブを停止します。パイプラインをドレインせずにキャンセルします。

  4. トピックを削除します。

    gcloud pubsub topics delete cron-topic

  5. パイプラインによって作成されたファイルを削除します。

    gsutil -m rm -rf "gs://${BUCKET_NAME}/samples/output*"

  6. Cloud Storage バケットを削除します。

    gsutil rb gs://${BUCKET_NAME}

次のステップ