Dataflow パイプラインの開発とテスト

このページでは、Dataflow パイプラインの開発とテストのベストプラクティスについて説明します。

概要

パイプラインのコードの実装方法は、本番環境でのパイプラインのパフォーマンスに大きく影響します。適正かつ効率的に機能するパイプラインコードを作成できるように、このドキュメントでは次のことについて説明します。

開発とデプロイのさまざまな段階でコードの実行をサポートするパイプラインランナー。
デプロイ環境。開発、テスト、本番前環境、本番環境でパイプラインを実行できます。
オープンソースのパイプラインコードとテンプレート。そのまま使用することも、コードを迅速に開発するために新しいパイプラインのベースとして使用することもできます。
パイプラインコードをテストするためのベストプラクティス。まず、このドキュメントでは、単体テスト、統合テスト、エンドツーエンドテストなどのテストの範囲とその関係などの概要について説明します。次に、テストデータを作成して統合する方法と各テストで使用するパイプラインランナーについて詳しく説明します。

パイプラインランナー

開発とテストでは、さまざまな Apache Beam ランナーを使用してパイプラインコードを実行します。Apache Beam SDK には、ローカルの開発用とテスト用に Direct Runner が用意されています。単体テストと統合テストでは、リリース自動化ツールを使って Direct Runner を使用することもできます。たとえば、継続的インテグレーション（CI）パイプラインで Direct Runner を使用できます。

Dataflow にデプロイされるパイプラインでは、Dataflow Runner を使用します。これにより、本番環境に似た環境でパイプラインが実行されます。また、アドホック開発のテストやエンドツーエンドのパイプラインテストにも Dataflow Runner を使用できます。

このページでは、Apache Beam Java SDK を使用してビルドされたパイプラインの実行に焦点を合わせていますが、Dataflow は Python と Go を使用して開発された Apache Beam パイプラインもサポートしています。Apache Beam Java、Python、Go SDK が Dataflow で一般提供されています。SQL デベロッパーは Apache Beam SQL を使用し、使い慣れた SQL 言語でパイプラインを作成することもできます。

デプロイ環境を設定する

開発のステージごとにユーザー、データ、コード、その他のリソースを分離するには、デプロイ環境を作成します。パイプライン開発のさまざまなステージで分離された環境を用意するときに、可能であれば、別々の Google Cloud プロジェクトを使用してください。

以降のセクションでは、一般的なデプロイ環境のセットについて説明します。

ローカル環境

ローカル環境は、デベロッパーワークステーションです。開発と迅速なテストを行うには、Direct Runner を使用してパイプラインコードをローカルで実行します。

Direct Runner を使用してローカルで実行されるパイプラインは、Pub/Sub トピックや BigQuery テーブルなどのリモート Google Cloud リソースとやり取りできます。個々のデベロッパーに個別の Google Cloud プロジェクトを用意すると、Google Cloud サービスでアドホックテストを行うためのサンドボックスを使用できます。

Pub/Sub や Bigtable などの一部の Google Cloud サービスでは、ローカル開発用のエミュレータが用意されています。Direct Runner でこれらのエミュレータを使用すると、ローカルでエンドツーエンドの開発とテストを行うことができます。

サンドボックス環境

サンドボックス環境は、デベロッパーがコード開発時に Google Cloud サービスにアクセスできるようにする Google Cloud プロジェクトです。パイプラインのデベロッパーは、Google Cloud プロジェクトを他のデベロッパーと共有したり、各自の独立したプロジェクトを使用したりできます。独立したプロジェクトを使用することで、共有リソースの使用量と割り当ての管理に関する計画の複雑さを軽減できます。

デベロッパーは、サンドボックス環境を使用して、Dataflow Runner を使用したアドホックパイプラインを実行します。サンドボックス環境は、コード開発フェーズで本番環境のランナーに対してコードのデバッグとテストを行う場合に役立ちます。たとえば、アドホックパイプラインを実行すると、デベロッパーは次のことが可能になります。

コード変更によるスケーリングの動作への影響を確認する。
Direct Runner と Dataflow Runner で考えられる動作の違いを把握する。
Dataflow でグラフの最適化がどのように行われるのかを理解する。

アドホックテストでは、デベロッパーはサンドボックス環境で Dataflow を実行するために、ローカル環境からコードをデプロイできます。

本番前環境

本番前環境は、エンドツーエンドテストのように本番環境に相当する条件が求められる開発フェーズに使用する環境です。本番前環境用に別のプロジェクトを使用します。可能であれば、本番環境と同様に構成します。同様に、本番環境と同様の規模でエンドツーエンドのテストを実施する場合は、Dataflow やその他のサービスの Google Cloud プロジェクトの割り当てを本番環境とできるだけ近いものにします。

必要であれば、本番前環境をさらに複数の環境に分割することもできます。たとえば、品質管理環境は、さまざまなワークロード条件でデータの正確性、鮮度、パフォーマンスなどのサービスレベル目標（SLO）をテストする品質アナリストの作業をサポートします。

エンドツーエンドテストでは、テスト範囲内でデータソースとシンクを統合します。これらを本番前環境で使用可能にする方法を検討してください。テストデータは本番前環境で保存できます。たとえば、テストデータは入力データと一緒に Cloud Storage バケットに保存されます。その他のケースでは、本番環境の別のサブスクリプションを介して Pub/Sub トピックなどの本番前環境以外からテストデータを受け取る場合があります。ストリーミングパイプラインの場合は、Dataflow Streaming Data Generator などを使用し、生成されたデータでエンドツーエンドテストを実行して、本番環境に似たデータ特性とボリュームをエミュレートすることもできます。

ストリーミングパイプラインの場合は、本番環境に変更を加える前に、本番前環境でパイプラインの更新をテストします。特に、ダウンタイムを回避するために並列パイプラインを実行するなど、複数のステップを調整する必要がある場合は、ストリーミングパイプラインの更新手順をテストして検証することが重要です。

本番環境

本番環境は専用の Google Cloud プロジェクトです。継続的デリバリーは、すべてのエンドツーエンドテストに合格すると、本番環境にデプロイアーティファクトをコピーします。

開発に関するベストプラクティス

Dataflow パイプラインのベストプラクティスをご覧ください。

パイプラインをテストする

ソフトウェア開発、単体テスト、統合テスト、エンドツーエンドテストは、ソフトウェアテストの一般的な種類です。これらのテストの種類は、データパイプラインにも適用できます。

Apache Beam SDK には、これらのテストを有効にする機能があります。テストのタイプごとに異なるデプロイ環境が理想的です。次の図は、単体テスト、統合テスト、エンドツーエンドテストがパイプラインとデータのさまざまな部分でどのように適用されるかを示しています。

テストの種類と、変換、パイプライン、データソース、データシンクとの関係

この図は、さまざまなテストの範囲と、それらの変換（DoFn と PTransform サブクラス）、パイプライン、データソース、データシンクとの関係を示しています。

次のセクションで、Dataflow を使用して一般的なソフトウェアテストをデータパイプラインに適用する方法を説明します。このセクションを読み終えたら、この図に戻り、さまざまな種類のテストがどのように関係しているかを確認してください。

データサンプリング

Dataflow パイプラインの各ステップでデータをモニタリングするには、テスト中のデータサンプリングを有効にします。これにより、変換の出力を表示して、出力が正しいことを確認できます。

単体テスト

単体テストでは、変換の出力と検証済みのデータ入力および出力のセットを比較して、DoFn サブクラスと複合変換（PTransform サブクラス）が正しく機能しているかどうか評価します。通常、デベロッパーはこれらのテストをローカル環境で実行できます。ビルド環境で継続的インテグレーション（CI）の単体テストの自動化を利用して、テストを自動的に実行することもできます。

Direct Runner は、変換のビジネスロジックのテストに焦点を当てた、参照用のテストデータのサブセットを使用して単体テストを実行します。テストデータは、テストを実行するマシンのローカルメモリに収まるサイズでなければなりません。

Apache Beam SDK には、個々の変換（DoFn サブクラス）、複合変換（PTransform サブクラス）、パイプライン全体に対して単体テストを実行するための TestPipeline という JUnit ルールが用意されています。JUnit テストクラスの次のコードスニペットに示されているように、Direct Runner や Dataflow Runner などの Apache Beam パイプラインランナー上で TestPipeline を使用し、PAssert を使用して PCollection オブジェクトのコンテンツにアサーションを適用できます。

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void myPipelineTest() throws Exception {
  final PCollection<String> pcol = p.apply(...)
  PAssert.that(pcol).containsInAnyOrder(...);
  p.run();
}

個別の変換の単体テスト

コードを再利用可能な変換（たとえば、トップレベルまたは静的なネストされたクラスとして）に分解することで、パイプラインのさまざまな部分を対象とするテストを作成できます。再利用可能な変換にはテスト以外のメリットもあります。パイプラインのビジネスロジックをコンポーネント部分にカプセル化することで、コードの保守性と再利用性を高めることができます。対照的に、パイプラインで匿名の個々の内部クラスを使用して変換を実装している場合は、パイプラインの個々の部分のテストが難しくなることがあります。

次の Java スニペットでは匿名の内部クラスとして変換が実装されているため、テストを簡単に行うことができません。

PipelineOptions options = PipelineOptionsFactory.create();

Pipeline p = Pipeline.create(options)

PCollection<Integer> output =
    p.apply("Read from text", TextIO.Read.from(...))
        .apply("Split words", ParDo.of(new DoFn() {
          // Untestable anonymous transform 1
        }))
        .apply("Generate anagrams", ParDo.of(new DoFn() {
          // Untestable anonymous transform 2
        }))
        .apply("Count words", Count.perElement());

次の例では、匿名の内部クラスが名前付きの具体的な DoFn サブクラスにリファクタリングされています。前の例と比べてみてください。エンドツーエンドのパイプラインを構成する具体的な DoFn サブクラスごとに、個別の単体テストを作成できます。

PipelineOptions options = PipelineOptionsFactory.create();

Pipeline p = Pipeline.create(options)

PCollection<Integer> output =
    p.apply("Read from text", TextIO.Read.from(...))
        .apply("Split words", ParDo.of(new SplitIntoWordsFn()))
        .apply("Generate anagrams", ParDo.of(new GenerateAnagramsFn()))
        .apply("Count words", Count.perElement());

各 DoFn サブクラスのテストは、単一変換を含むバッチパイプラインの単体テストに似ています。Create 変換を使用してテストデータの PCollection オブジェクトを作成し、DoFn オブジェクトに渡します。PAssert を使用して、PCollection オブジェクトのコンテンツが正しいことを確認します。次の Java コードの例では、正しい出力形式を確認するために PAssert クラスを使用しています。

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void testGenerateAnagramsFn() {
    // Create the test input
    PCollection<String> words = p.apply(Create.of("friend"));

    // Test a single DoFn using the test input
    PCollection<String> anagrams =
        words.apply("Generate anagrams", ParDo.of(new GenerateAnagramsFn()));

    // Assert correct output from
    PAssert.that(anagrams).containsInAnyOrder(
        "finder", "friend", "redfin", "refind");

    p.run();
}

統合テスト

統合テストでは、パイプライン全体が正しく機能していることを確認します。次のような統合テストについて考えてみましょう。

データパイプラインを構成する個々の変換が統合された機能を評価する変換統合テスト。変換統合テストは、外部のデータソースとデータシンクとの統合を除けば、パイプライン全体の単体テストと考えることができます。Apache Beam SDK には、データパイプラインにテストデータを提供し、処理の結果を検証するメソッドが用意されています。変換統合テストを実行するために Direct Runner が使用されます。
データパイプラインのライブデータソースとデータシンクとの統合を評価するシステム統合テスト。パイプラインが外部システムと通信するには、外部サービスにアクセスするための適切な認証情報を使用してテストを構成する必要があります。ストリーミングパイプラインは無期限に実行されるため、実行中のパイプラインをいつ、どのように停止するかを決めておく必要があります。Direct Runner を使用してシステム統合テストを実行することで、Dataflow ジョブを送信して完了を待つ必要がなくなり、パイプラインと他のシステムの統合をすばやく検証できます。

注: システム統合テストの代わりに小規模なエンドツーエンドテストを実行することがよくあります。エンドツーエンドテストでは、Dataflow Runner と小規模なテストデータセットを使用して、データソース、データシンク、その他の外部システムとの統合など、パイプライン全体が正しく機能することを検証します。

デベロッパーの生産性を低下させずに欠陥を迅速に検出してフィードバックを提供するように、変換とシステム統合テストを設計する必要があります。Dataflow ジョブとして実行されるテストなど、実行時間の長いテストの場合は、実行頻度の低いエンドツーエンドテストが適している可能性があります。

データパイプラインは 1 つ以上の関連する変換と考えることができます。パイプラインのカプセル化複合変換を作成し、TestPipeline を使用してパイプライン全体の統合テストを実行できます。パイプラインのテストモード（バッチモードまたはストリーミングモード）によって、Create 変換または TestStream 変換のどちらかを使用してテストデータを提供します。

統合テストにテストデータを使用する

本番環境では、パイプラインがさまざまなデータソースやデータシンクと統合されている可能性があります。しかし、単体テストと変換統合テストでは、テスト入力を提供して出力を直接検証することで、パイプラインコードのビジネスロジックの検証に焦点を合わせる必要があります。このアプローチにより、テストの簡素化だけでなく、パイプライン固有の問題とデータソースやデータシンクに起因する可能性のある問題を切り分けることができます。

バッチパイプラインをテストする

バッチパイプラインでは、Create 変換を使用して、標準のインメモリコレクション（Java List オブジェクトなど）から入力テストデータの PCollection オブジェクトを作成します。テストデータが小さく、コードに含めることができる場合は、Create 変換を使用します。その後、出力の PCollection オブジェクトに PAssert を使用して、パイプラインコードが正しいかどうかを確認できます。この方法は、Direct Runner と Dataflow Runner でサポートされています。

次の Java コードスニペットは、出力の PCollection オブジェクトに対するパイプライン（WeatherStatsPipeline）を構成する個々の変換の一部、または全部を含む複合変換のアサーションを示しています。このアプローチは、パイプラインの個々の変換の単体テストに似ています。

private class WeatherStatsPipeline extends
    PTransform<PCollection<Integer>, PCollection<WeatherSummary>> {
  @Override
  public PCollection<WeatherSummary> expand(PCollection<Integer> input) {
    // Pipeline transforms …
  }
}

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void testWeatherPipeline() {
  // Create test input consisting of temperature readings
  PCollection<Integer> tempCelsius =
      p.apply(Create.of(24, 22, 20, 22, 21, 21, 20));

  // CalculateWeatherStats calculates the min, max, and average temperature
  PCollection<WeatherSummary> result =
      tempCelsius.apply("Calculate weather statistics", new WeatherStatsPipeline());

   // Assert correct output from CalculateWeatherStats
   PAssert.thatSingleton(result).isEqualTo(new WeatherSummary.Builder()
       .withAverageTemp(21)
       .withMaxTemp(24)
       .withMinTemp(20)
       .build());

   p.run();
}

ウィンドウ処理の動作をテストするには、次のコードスニペットのように、Create 変換を使用してタイムスタンプ付きの要素を作成することもできます。

private static final Duration WINDOW_DURATION = Duration.standardMinutes(3);

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void testWindowedData() {
    PCollection<String> input =
        p.apply(
            Create.timestamped(
                    TimestampedValue.of("a", new Instant(0L)),
                    TimestampedValue.of("a", new Instant(0L)),
                    TimestampedValue.of("b", new Instant(0L)),
                    TimestampedValue.of("c", new Instant(0L)),
                    TimestampedValue.of("c", new Instant(0L).plus(WINDOW_DURATION)))
                .withCoder(StringUtf8Coder.of()));

   PCollection<KV<String, Long>> windowedCount =
       input
           .apply(Window.into(FixedWindows.of(WINDOW_DURATION)))
           .apply(Count.perElement());

    PAssert.that(windowedCount)
        .containsInAnyOrder(
            // Output from first window
            KV.of("a", 2L),
            KV.of("b", 1L),
            KV.of("c", 1L),
            // Output from second window
            KV.of("c", 1L));

   p.run();
}

ストリーミングパイプラインをテストする

ストリーミングパイプラインには、制限なしのデータの処理方法を定義する前提条件が存在します。こうした前提条件は多くの場合、実際の状況でのデータに適時性に関するものであるため、前提条件が true か false かに応じて正確性に影響を及ぼします。したがって、ストリーミングパイプラインの統合テストでは、ストリーミングデータの到着に関する非決定的な特性をシミュレートするテストを含めるのが理想的です。

このようなテストを可能にするため、Apache Beam SDK には、データパイプラインの結果で要素のタイミング（早い、時間どおり、遅い）の効果をモデル化する TestStream クラスが用意されています。これらのテストを PAssert クラスとともに使用して、想定される結果と比較します。

TestStream は、Direct Runner と Dataflow Runner でサポートされています。次のコードサンプルでは、TestStream 変換を作成します。

final Duration WINDOW_DURATION = Duration.standardMinutes(3);

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void testDroppedLateData() {
   TestStream<String> input = TestStream.create(StringUtf8Coder.of())
      // Add elements arriving before the watermark
      .addElements(
         TimestampedValue.of("a", new Instant(0L)),
         TimestampedValue.of("a", new Instant(0L)),
         TimestampedValue.of("b", new Instant(0L)),
         TimestampedValue.of("c", new Instant(0L).plus(Duration.standardMinutes(3))))
         // Advance the watermark past the end of the window
      .advanceWatermarkTo(new Instant(0L).plus(WINDOW_DURATION).plus(Duration.standardMinutes(1)))
      // Add elements which will be dropped due to lateness
      .addElements(
         TimestampedValue.of("c", new Instant(0L)))
      // Advance the watermark to infinity which will close all windows
      .advanceWatermarkToInfinity();

      PCollection<KV<String, Long>> windowedCount =
          p.apply(input)
             .apply(Window.into(FixedWindows.of(WINDOW_DURATION)))
             .apply(Count.perElement());

   PAssert.that(windowedCount)
      .containsInAnyOrder(
          // Output from first window
          KV.of("a", 2L),
          KV.of("b", 1L),
          KV.of("c", 1L));

   p.run();
}

TestStream の詳細については、Apache Beam での制限なしのパイプラインのテストをご覧ください。単体テストで Apache Beam SDK を使用する方法については、Apache Beam のドキュメントをご覧ください。

統合テストで Google Cloud サービスを使用する

Direct Runner は Google Cloud サービスと統合できるため、必要に応じてローカル環境のアドホックテストやシステム統合テストで Pub/Sub、BigQuery などのサービスを使用できます。Direct Runner を使用する場合、パイプラインは、gcloud コマンドラインツールで構成されたユーザーアカウント、または GOOGLE_APPLICATION_CREDENTIALS 環境変数に指定したサービスアカウントとして実行されます。したがって、パイプラインを実行する前に、必要なすべてのリソースに対する十分な権限をこのアカウントに付与する必要があります。詳細については、Dataflow のセキュリティと権限をご覧ください。

完全にローカルな統合テストでは、一部の Google Cloud サービスでローカルエミュレータを使用できます。ローカルエミュレータは Pub/Sub と Bigtable で使用できます。

ストリーミングパイプラインのシステム統合テストでは、setBlockOnRun メソッド（DirectOptions インターフェースに定義）を使用して、Direct Runner でパイプラインを非同期で実行できます。それ以外の場合は、パイプラインが手動で終了されるまで、パイプラインの実行で呼び出し元の親プロセス（ビルドパイプラインのスクリプトなど）がブロックされます。パイプラインを非同期で実行する場合は、次のコード例のように、返された PipelineResult インスタンスを使用してパイプラインの実行をキャンセルできます。

public interface StreamingIntegrationTestOptions extends
   DirectOptions, StreamingOptions, MyOtherPipelineOptions {
   ...
}

@Rule
public final transient TestPipeline p = TestPipeline.create();

@Test
@Category(NeedsRunner.class)
public void testNonBlockingPipeline() {
    StreamingIntegrationTestOptions options =
        p.getOptions().as(StreamingIntegrationOptions.class);

    options.setBlockOnRun(false); // Set non-blocking pipeline execution
    options.setStreaming(true); // Set streaming mode

    p.apply(...); // Apply pipeline transformations

    PipelineResult result = p.run(); // Run the pipeline

    // Generate input, verify output, etc
    ...

    // Later on, cancel the pipeline using the previously returned
    result.cancel();
}

エンドツーエンドテスト

エンドツーエンドテストでは、本番環境とよく似ている条件において、Dataflow Runner でエンドツーエンドパイプラインを実行して、オペレーションが正しく行われたことを検証します。このテストでは、Dataflow Runner を使用したビジネスロジックが正しく機能することや、本番環境相当の負荷でパイプラインが期待どおりに動作するかどうかを確認します。通常、エンドツーエンドテストは、本番前環境として指定された専用の Google Cloud プロジェクトで実行します。

さまざまなスケールでパイプラインをテストするには、さまざまなタイプのエンドツーエンドテストを使用します。次に例を示します。

テストデータセットの一部分（全体の 1% 程度）を使用して、小規模なエンドツーエンドテストを実行し、本番前環境でパイプライン機能をすばやく検証します。
完全なテストデータセットを使用して、大規模なエンドツーエンドテストを実行し、本番環境相当のデータ量と条件でパイプラインの機能を検証します。

ストリーミングパイプラインで同じデータを使用できる場合は、本番環境パイプラインと並行してテストパイプラインを実行することをおすすめします。これにより、自動スケーリングやパフォーマンスなどの結果とオペレーションの動作を比較できます。

エンドツーエンドテストは、パイプラインが本番環境の SLO をどの程度満たすかを予測するのに役立ちます。本番前環境は、本番環境と同様の条件でパイプラインをテストします。エンドツーエンドテストでは、Dataflow Runner を使用してパイプラインを実行し、本番環境のデータセットと一致する（またはよく似た）完全な参照データセットを処理します。

実際のデータを正確にシミュレートするテスト用の合成データを生成できない場合があります。この問題に対処するためのアプローチの一つは、本番環境のデータソースからの抽出をクレンジングして参照データセットを作成することです。ここでは、機密データが適切な変換によって匿名化されます。この目的のため、機密データ保護を使用することをおすすめします。機密データ保護は、さまざまなコンテンツタイプとデータソースから機密データを検出し、秘匿化、マスキング、フォーマット保持暗号化、日付シフトなど、さまざまな匿名化手法を適用できます。

バッチパイプラインとストリーミングパイプラインのエンドツーエンドテストの違い

大規模なテストデータセットに完全なエンドツーエンドテストを実行する前に、テストデータの一部（全体の 1% 程度）を使用してテストを行い、短時間で期待される動作を確認することをおすすめします。Direct Runner を使用する統合テストと同様に、Dataflow Runner を使用してパイプラインを実行するときに PCollection オブジェクトに PAssert を使用できます。PAssert の詳細については、このページの単体テストのセクションをご覧ください。

ユースケースによっては、エンドツーエンドテストで非常に大規模な出力を検証するのは、実用的でなかったり、費用がかかる場合があります。その場合は、代わりに出力結果セットからの代表サンプルで検証できます。たとえば、BigQuery を使用して出力行をサンプリングし、期待される結果の参照データセットと比較できます。

ストリーミングパイプラインでは、合成データでの現実的なストリーミング条件のシミュレーションが困難な場合があります。エンドツーエンドテストにストリーミングデータを提供する一般的な方法は、本番環境のデータソースとテストを統合することです。Pub/Sub をデータソースとして使用する場合、既存のトピックへの追加のサブスクリプションにより、エンドツーエンドテスト用に別のデータストリームを有効にできます。次に、同じデータを使用する異なるパイプラインの結果を比較できます。これは、他の本番前環境および本番環境のパイプラインに対してパイプラインの候補を検証する際に役立ちます。

次の図は、この方法によって本番環境パイプラインとテストパイプラインを異なるデプロイ環境で並行して実行する方法を示しています。

単一の Pub/Sub ストリーミングソースを使用した本番環境パイプラインと並行したテストパイプラインの実行

この図では、両方のパイプラインが同じ Pub/Sub トピックから読み取りますが、個別のサブスクリプションを使用します。この設定により、2 つのパイプラインが同じデータを独立して処理するため、結果を比較できます。テストパイプラインは本番環境プロジェクトとは別のサービスアカウントを使用するため、本番環境プロジェクトの Pub/Sub サブスクライバーの割り当ては使用されません。

バッチパイプラインとは異なり、ストリーミングパイプラインは明示的にキャンセルされるまで継続されます。エンドツーエンドテストでは、パイプラインをある時期まで（たとえばエンドツーエンドテストの実行まで）そのまま実行するか、結果を確認できるようテストの完了を表すポイントでパイプラインをキャンセルするかを決定する必要があります。

使用するテストデータの種類は、この決定に影響します。たとえば、ストリーミングパイプラインに渡される制限付きのテストデータのセットを使用する場合、すべての要素の処理が完了したらパイプラインをキャンセルできます。あるいは、現実のデータソース（本番環境で使用されている既存の Pub/Sub トピックなど）を使用する場合や、継続的にテストデータを別に生成している場合は、より長い期間テストパイプラインの実行を継続することをおすすめします。後者の場合、本番環境での挙動や他のテストパイプラインとの動作を比較できます。