Dataflow を使用して Pub/Sub Lite メッセージをストリーミングする

Dataflow を使用して Pub/Sub Lite メッセージをストリーミングする

独自のデータ処理プログラムを作成して実行する代わりに、Dataflow を Apache Beam の Pub/Sub Lite I/O コネクタとともに使用することもできます。Dataflow は、ストリーミング（リアルタイム）モードのデータとバッチモードのデータを同等の信頼性と明瞭度で変換、活用するフルマネージド サービスです。Dataflow は、Apache Beam SDK を使用して開発されたプログラムを確実に実行します。Apache Beam SDK は、強力なステートフル処理抽象化と、他のストリーミング システムとバッチシステムへの I/O コネクタで構成された拡張可能なセットです。

このクイックスタートでは、以下を行う Apache Beam パイプラインを作成する方法を示します。

• Pub/Sub Lite からメッセージを読み取る
• タイムスタンプごとにメッセージをウィンドウ処理（グループ化）する
• Cloud Storage にメッセージを書き込む

また、次の方法についても説明します。

• Dataflow で実行するため、パイプラインを送信する
• パイプラインから Dataflow フレックス テンプレートを作成する

このチュートリアルでは Maven が必要ですが、サンプル プロジェクトを Maven から Gradle に変換することもできます。詳細については、省略可: Maven から Gradle に変換するをご覧ください。

始める前に

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

3. gcloud CLI を初期化するには:

   gcloud init

4. Google Cloud プロジェクトを作成または選択します。

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

   • Google Cloud プロジェクトを作成します。

     gcloud projects create PROJECT_ID

     PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。

   • 作成した Google Cloud プロジェクトを選択します。

     gcloud config set project PROJECT_ID

     PROJECT_ID は、実際の Google Cloud プロジェクト名に置き換えます。

5. Google Cloud プロジェクトで課金が有効になっていることを確認します。

6. Pub/Sub Lite, Dataflow, Google Cloud Storage JSON API, and Cloud Logging API を有効にします。

   gcloud services enable pubsublite.googleapis.com dataflow.googleapis.com storage-api.googleapis.com logging.googleapis.com

7. 認証を設定します。

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

      gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

      SERVICE_ACCOUNT_NAME をサービス アカウントの名前に置き換えます。

   2. サービス アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。roles/dataflow.worker, roles/storage.objectAdmin, roles/pubsublite.admin

      gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=ROLE

      次のように置き換えます。

      • SERVICE_ACCOUNT_NAME: サービス アカウントの名前
      • PROJECT_ID: サービス アカウントを作成したプロジェクト ID
      • ROLE: 付与するロール
      注: --role フラグは、サービス アカウントがプロジェクト内のどのリソースにアクセスできるかに影響します。これらのロールは後で取り消すことも、追加のロールを付与することもできます。本番環境では、オーナー、編集者、閲覧者のロールを付与しないでください。代わりに、ニーズに合わせて事前定義ロールまたはカスタムロールを付与します。

   3. サービス アカウントのロールを使用して、そのサービス アカウントを他のリソースに関連付けることができるロールを Google アカウントに付与します。

      gcloud iam service-accounts add-iam-policy-binding SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com --member="user:USER_EMAIL" --role=roles/iam.serviceAccountUser

      次のように置き換えます。

      • SERVICE_ACCOUNT_NAME: サービス アカウントの名前
      • PROJECT_ID: サービス アカウントを作成したプロジェクト ID
      • USER_EMAIL: Google アカウントのメールアドレス
8. Google Cloud CLI をインストールします。

9. gcloud CLI を初期化するには:

   gcloud init

10. Google Cloud プロジェクトを作成または選択します。

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

    • Google Cloud プロジェクトを作成します。

      gcloud projects create PROJECT_ID

      PROJECT_ID は、作成する Google Cloud プロジェクトの名前に置き換えます。

    • 作成した Google Cloud プロジェクトを選択します。

      gcloud config set project PROJECT_ID

      PROJECT_ID は、実際の Google Cloud プロジェクト名に置き換えます。

11. Google Cloud プロジェクトで課金が有効になっていることを確認します。

12. Pub/Sub Lite, Dataflow, Google Cloud Storage JSON API, and Cloud Logging API を有効にします。

    gcloud services enable pubsublite.googleapis.com dataflow.googleapis.com storage-api.googleapis.com logging.googleapis.com

13. 認証を設定します。

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

       gcloud iam service-accounts create SERVICE_ACCOUNT_NAME

       SERVICE_ACCOUNT_NAME をサービス アカウントの名前に置き換えます。

    2. サービス アカウントにロールを付与します。次の IAM ロールごとに次のコマンドを 1 回実行します。roles/dataflow.worker, roles/storage.objectAdmin, roles/pubsublite.admin

       gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com" --role=ROLE

       次のように置き換えます。

       • SERVICE_ACCOUNT_NAME: サービス アカウントの名前
       • PROJECT_ID: サービス アカウントを作成したプロジェクト ID
       • ROLE: 付与するロール
       注: --role フラグは、サービス アカウントがプロジェクト内のどのリソースにアクセスできるかに影響します。これらのロールは後で取り消すことも、追加のロールを付与することもできます。本番環境では、オーナー、編集者、閲覧者のロールを付与しないでください。代わりに、ニーズに合わせて事前定義ロールまたはカスタムロールを付与します。

    3. サービス アカウントのロールを使用して、そのサービス アカウントを他のリソースに関連付けることができるロールを Google アカウントに付与します。

       gcloud iam service-accounts add-iam-policy-binding SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com --member="user:USER_EMAIL" --role=roles/iam.serviceAccountUser

       次のように置き換えます。

       • SERVICE_ACCOUNT_NAME: サービス アカウントの名前
       • PROJECT_ID: サービス アカウントを作成したプロジェクト ID
       • USER_EMAIL: Google アカウントのメールアドレス

14. Google アカウントのローカル認証情報を作成します。

    gcloud auth application-default login

Pub/Sub Lite プロジェクトを設定する

1. Cloud Storage バケット、プロジェクト、Dataflow リージョンの変数を作成します。Cloud Storage のバケット名は、グローバルに一意にする必要があります。Dataflow リージョンは、ジョブを実行できる有効なリージョンである必要があります。リージョンとロケーションの詳細については、Dataflow のロケーションをご覧ください。

   export PROJECT_ID=$(gcloud config get-value project)
   export SERVICE_ACCOUNT=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com
   export BUCKET=BUCKET_NAME
   export DATAFLOW_REGION=DATAFLOW_REGION
   

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

      gsutil mb gs://$BUCKET
   

Pub/Sub Lite ゾーン Lite のトピックとサブスクリプションを作成する

ゾーン Lite Pub/Sub Lite トピックと Lite サブスクリプションを作成します。

Lite ロケーションの場合は、サポートされている Pub/Sub Lite のロケーションを選択します。また、リージョンのゾーンも指定する必要があります。例: us-central1-a。

export TOPIC=LITE_TOPIC_ID
export SUBSCRIPTION=LITE_SUBSCRIPTION_ID
export LITE_LOCATION=LITE_LOCATION
gcloud pubsub lite-topics create $TOPIC \
      --location=$LITE_LOCATION \
      --partitions=1 \
      --per-partition-bytes=30GiB
gcloud pubsub lite-subscriptions create $SUBSCRIPTION \
      --location=$LITE_LOCATION \
      --topic=$TOPIC \
      --starting-offset=beginning

Dataflow にメッセージをストリーミングする

クイックスタートのサンプルコードをダウンロードする

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

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

サンプルコード

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

• 制限なしソースとして Pub/Sub Lite サブスクリプションからメッセージを読み取る。
• 固定時間ウィンドウとデフォルト トリガーを使用して、パブリッシュ タイムスタンプに基づいてメッセージをグループ化します。

• グループ化したメッセージを Cloud Storage のファイルに書き込む。

Java

このサンプルを実行する前に、Pub/Sub Lite クライアント ライブラリの Java の設定手順を実施してください。

import com.google.cloud.pubsublite.SubscriptionPath;
import com.google.cloud.pubsublite.proto.SequencedMessage;
import org.apache.beam.examples.common.WriteOneFilePerWindow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.pubsublite.PubsubLiteIO;
import org.apache.beam.sdk.io.gcp.pubsublite.SubscriberOptions;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
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.transforms.MapElements;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PubsubliteToGcs {
  /*
   * Define your own configuration options. Add your arguments to be processed
   * by the command-line parser.
   */
  public interface PubsubliteToGcsOptions extends StreamingOptions {
    @Description("Your Pub/Sub Lite subscription.")
    @Required
    String getSubscription();

    void setSubscription(String value);

    @Description("Window size of output files in minutes.")
    @Default.Integer(1)
    Integer getWindowSize();

    void setWindowSize(Integer value);

    @Description("Filename prefix of output files.")
    @Required
    String getOutput();

    void setOutput(String value);
  }

  private static final Logger LOG = LoggerFactory.getLogger(PubsubliteToGcs.class);

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

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

    options.setStreaming(true);

    SubscriberOptions subscriberOptions =
        SubscriberOptions.newBuilder()
            .setSubscriptionPath(SubscriptionPath.parse(options.getSubscription()))
            .build();

    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply("Read From Pub/Sub Lite", PubsubLiteIO.read(subscriberOptions))
        .apply(
            "Convert messages",
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (SequencedMessage sequencedMessage) -> {
                      String data = sequencedMessage.getMessage().getData().toStringUtf8();
                      LOG.info("Received: " + data);
                      long publishTime = sequencedMessage.getPublishTime().getSeconds();
                      return data + "\t" + publishTime;
                    }))
        .apply(
            "Apply windowing function",
            Window
                // Group the elements using fixed-sized time intervals based on the element
                // timestamp (using the default event time trigger). The element timestamp
                // is the publish timestamp associated with a message.
                //
                // NOTE: If data is not being continuously ingested, such as with a batch or
                // intermittent publisher, the final window will never close as the watermark
                // will not advance. If this is a possibility with your pipeline, you should
                // add an additional processing time trigger to force window closure after
                // enough time has passed. See
                // https://beam.apache.org/documentation/programming-guide/#triggers
                // for more information.
                .<String>into(FixedWindows.of(Duration.standardMinutes(options.getWindowSize()))))
        .apply("Write elements to GCS", new WriteOneFilePerWindow(options.getOutput(), numShards));

    // Execute the pipeline. You may add `.waitUntilFinish()` to observe logs in your console, but
    // `waitUntilFinish()` will not work in Dataflow Flex Templates.
    pipeline.run();
  }
}

Dataflow パイプラインを開始する

Dataflow でパイプラインを開始するには、次のコマンドを実行します。

mvn compile exec:java \
    -Dexec.mainClass=examples.PubsubliteToGcs \
    -Dexec.args=" \
        --subscription=projects/$PROJECT_ID/locations/$LITE_LOCATION/subscriptions/$SUBSCRIPTION \
        --output=gs://$BUCKET/samples/output \
        --windowSize=1 \
        --project=$PROJECT_ID \
        --region=$DATAFLOW_REGION \
        --tempLocation=gs://$BUCKET/temp \
        --runner=DataflowRunner \
        --serviceAccount=$SERVICE_ACCOUNT"

上記のコマンドは Dataflow ジョブを起動します。コンソール出力のリンクをたどって、Dataflow モニタリング コンソールでジョブにアクセスします。

ジョブの進行状況を確認する

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

Dataflow コンソールに移動

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

• ジョブグラフ
• 実行の詳細
• ジョブの指標

Lite トピックにメッセージを公開します。

gcloud pubsub lite-topics publish $TOPIC \
    --location=$LITE_LOCATION \
    --message="Hello World!"

ワーカーログにメッセージが表示されるまで、数分間かかる場合があります。

次のコマンドを使用して、Cloud Storage に書き込まれたファイルを確認します。

gsutil ls "gs://$BUCKET/samples/"

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

 gs://$BUCKET/samples/output-19:41-19:42-0-of-1
 gs://$BUCKET/samples/output-19:47-19:48-0-of-1
 gs://$BUCKET/samples/output-19:48-19:49-0-of-1

次のコマンドを使用して、ファイルの内容を確認します。

gsutil cat "gs://$BUCKET/samples/your-filename"

省略可: Dataflow テンプレートを作成する

必要に応じて、パイプラインに基づいたカスタム Dataflow フレックス テンプレートを作成できます。Dataflow テンプレートを使用すると、完全な Java 開発環境をセットアップしなくても、Google Cloud コンソールやコマンドラインから異なる入力パラメータを使用してジョブを実行できます。

1. パイプラインのすべての依存関係を含むファット JAR を作成します。コマンドの実行後、target/pubsublite-streaming-bundled-1.0.jar が表示されます。

   mvn clean package -DskipTests=true
   

2. テンプレート ファイルとテンプレート コンテナ イメージの名前と場所を指定します。

   export TEMPLATE_PATH="gs://$BUCKET/samples/your-template-file.json"
   export TEMPLATE_IMAGE="gcr.io/$PROJECT_ID/your-template-image:latest"
   

3. カスタム Flex テンプレートを作成します。この例では、ジョブの実行に必要な仕様を含む metadata.json ファイルが指定されています。

   gcloud dataflow flex-template build $TEMPLATE_PATH \
       --image-gcr-path $TEMPLATE_IMAGE \
       --sdk-language "JAVA" \
       --flex-template-base-image "JAVA11" \
       --metadata-file "metadata.json" \
       --jar "target/pubsublite-streaming-bundled-1.0.jar" \
       --env FLEX_TEMPLATE_JAVA_MAIN_CLASS="examples.PubsubliteToGcs"
   

4. カスタム Flex テンプレートを使用してジョブを実行します。

   注: Pub/Sub Lite では、1 つのパーティションからメッセージを pull できるサブスクライバーは 1 人だけです。トピックが 1 つのパーティションのみを持つように構成されていて、そのサブスクリプションを複数の Dataflow ジョブで使用する場合、1 つのジョブのみがメッセージを受信します。

Console

1. テンプレートからジョブを作成.

2. ジョブ名を入力します。

3. Dataflow リージョンを入力します。

4. カスタム テンプレートを選択します。

5. テンプレート パスを入力します。

6. 必要なパラメータを入力します。

7. [ジョブを実行] をクリックします。

gcloud

gcloud dataflow flex-template run "pubsublite-to-gcs-`date +%Y%m%d`" \
     --template-file-gcs-location $TEMPLATE_PATH \
     --parameters subscription="projects/$PROJECT_ID/locations/$LITE_LOCATION/subscriptions/$SUBSCRIPTION" \
     --parameters output="gs://$BUCKET/samples/template-output" \
     --parameters windowSize=1 \
     --region $DATAFLOW_REGION \
     --serviceAccount=$SERVICE_ACCOUNT

クリーンアップ

このページで使用したリソースに対して Google Cloud アカウントで課金されないようにするには、Google Cloud プロジェクトとそのリソースを削除します。

1. Dataflow コンソールで、ジョブを停止します。パイプラインは、ドレインするのではなくキャンセルします。

2. トピックとサブスクリプションを削除します。

   gcloud pubsub lite-topics delete $TOPIC
   gcloud pubsub lite-subscriptions delete $SUBSCRIPTION
   

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

   gsutil -m rm -rf "gs://$BUCKET/samples/*"
   gsutil -m rm -rf "gs://$BUCKET/temp/*"
   

4. テンプレート イメージとテンプレート ファイルが存在する場合は、それらを削除します。

   gcloud container images delete $TEMPLATE_IMAGE
   gsutil rm $TEMPLATE_PATH
   

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

   gsutil rb gs://$BUCKET
   

6. サービス アカウントを削除します。

   gcloud iam service-accounts delete SERVICE_ACCOUNT_EMAIL

7. 作成した認証情報を取り消して、ローカル認証情報ファイルを削除します。

   gcloud auth application-default revoke

8. （省略可）gcloud CLI から認証情報を取り消します。

   gcloud auth revoke

次のステップ

• Dataflow フレックス テンプレートの構成の詳細を確認する。

• Dataflow ストリーミング パイプラインを理解する。