破壊的変更

ストリーミングパイプラインやインクリメンタル（差分処理）パイプラインでステートフル関数を変更すると、破壊的変更が発生します。データ加工には、行レベルとステートフルの2種類があります。

• 行レベルのデータ加工： 結果を生成するために必要なのは1行のデータのみです。たとえば、数値の乗算やフィルターが該当します。
• ステートフル関数： 結果を生成するために複数行のデータを必要とするデータ加工です。

主なステートフル関数は5つあります。

• 集計（ストリーミングではウィンドウ内の集計）
• 外部キャッシュ結合（ストリーミングのみ）
• ハートビート検出（ストリーミングのみ）
• 時間制限付き重複削除（ストリーミングのみ）
• 時間制限付きイベント時刻の並べ替え（ストリーミングのみ）

ステートフル関数を変更すると、それまでの出力が正確でなくなる可能性があります。たとえば、偶数に絞り込み、その集合の合計を計算しているとします。フィルターを変更してすべての奇数を対象にすると、既存の状態は偶数の合計ですが、新しく絞り込まれる値はすべて奇数になります。そのため、偶数の集合の合計に奇数の集合の合計を加えた値となり、この合計が何を表しているのかが曖昧になります。状態を更新するには、リプレイを実行できます。

リプレイには2種類あります。

• インプットデータの最初からリプレイ： データの最初からパイプラインをリプレイします。入力がストリームかインクリメンタルデータセットかに応じて、ストリームの開始時点、または入力データセットの最初のトランザクションからリプレイします。

• 指定した時間前からリプレイ（ストリーミングでのみ利用可能）： 指定した時間前からの上流データを使用してパイプラインをリプレイします。この詳細なリプレイには、指定した時刻より前にコミットされた最初のトランザクション以降のすべてのデータが含まれ、それより前のデータは処理されません。つまり、指定した時刻より前のデータが1トランザクション分含まれる可能性があります。

リプレイは任意の場合と必須の場合があります。破壊的変更の場合、Pipeline Builder はその変更を自動的に検出し、デプロイ時にリプレイを必須とします。以下の画像は、インクリメンタルパイプラインで強制されるリプレイを示しています。

状態の破壊が発生すると、Pipeline Builder はリプレイをエラータブに移動し、破壊が解消されるまでパイプラインのデプロイをブロックします。これにより、デプロイを進める前に、破壊的変更を了承して対処することが保証されます。

意図しないリプレイを防ぐため、パイプラインをデプロイパネルの詳細設定セクションはデフォルトで折りたたまれており、すべてのリプレイオプションが未選択になっています。リプレイ戦略を選択すると、リプレイ戦略の確認モーダルが表示され、実行前に戦略の説明と下流への影響を確認できます。

デプロイに状態を破壊する変更が含まれる場合、Pipeline Builder はデプロイタイムラインにリプレイ戦略の情報を表示します。破壊的変更が検出されたかどうかと、選択されたリプレイ戦略を確認できます。この情報はデプロイイベントの詳細にタスク行として表示されるため、デプロイ時にリプレイについてどのような判断が行われたかを把握できます。

状態の破壊の一般的な原因

以下の変更は、完全なリプレイを必要とする状態の破壊を引き起こすことがよくあります。これらのシナリオを理解することで、パイプラインの変更を計画し、予期しないダウンタイムを回避しやすくなります。

ステートフルなデータ加工の変更

ステートフルなデータ加工（集計、外部キャッシュ結合、ハートビート検出など）を削除または変更するには、完全なリプレイが必要です。これには、ステートフルなデータ加工に供給される入力の削除も含まれます。削除されたデータがないと、ステートフル関数が一貫した結果を生成できなくなるためです。

ステートフルなデータ加工に影響する入力の変更

入力を追加または削除すると、状態の破壊が発生します。この破壊は、完全なリプレイを行わずに了承できます。ただし、ステートフルなデータ加工に供給される入力を変更すると、過去のデータとの論理的な整合性が損なわれる可能性があります。たとえば、ステートフルな集計が特定の入力に依存していて、その入力が削除または置換された場合、既存の状態は、パイプラインが処理するデータを正確に反映しなくなります。

出力スキーマの変更

出力スキーマに新しい列を追加しても、リプレイは必要ありません。ただし、出力スキーマから列を削除すると、リプレイが必要な状態の破壊が発生します。既存の出力データに、更新されたスキーマに一致しなくなった列が含まれているためです。

ソースパイプラインの変更

現在のパイプラインにデータを供給するパイプラインをリプレイすると、現在のパイプラインもリプレイする必要が生じる場合があります。ソースデータが再処理されると、パイプラインの入力に届くデータが、パイプラインで最初に処理したデータと異なる可能性があり、ステートフルなデータ加工の現在の状態が無効になることがあります。

上流のストリーミングパイプラインがリプレイ時に出力をリセットを有効にして実行される場合（デフォルトの動作）、上流パイプラインの出力データセットはクリアされ、リプレイウィンドウ内のデータのみが再度格納されます。この場合、下流のパイプラインは特定の時間ウィンドウからではなく、入力データの最初からリプレイする必要があります。これにより、オントロジーのオブジェクトなどの下流の出力が、重複を作成することなく、再処理された期間全体にわたって正しく再構築されます。

状態を保持する変更

Pipeline Builder には、リプレイせずにパイプラインに特定の変更を加えられる機能があります。これらの機能により、ストリーム履歴を保持し、下流の利用側への影響を避けながら、前回の処理位置から処理を続行できます。

入力と出力の変更

デプロイ後にパイプラインの入力と出力を変更できます。動作は変更の種類によって異なります。

• 入力の追加： 新しい入力は最初から読み取られ、既存の入力は最後に処理した位置から続行されます。
• 入力または出力の削除： 削除した入力または出力に関連する状態は、リプレイを必要とせずに処理クラスターから削除されます。
• 出力の追加： 出力を追加すると状態の断絶が発生し、一定時間前または入力データの先頭からのリプレイが必要になります。

Pipeline Builder が入力または出力の変更を検出すると、状態の断絶モジュールに変更への了承を求めるプロンプトが表示されます。了承することで、リプレイを必要とせず、前回の処理位置から処理を続行するようシステムに指示します。

スキーマの変更

入力スキーマは、パイプラインのデプロイ時に固定されます。入力スキーマが変更されても、手動で再デプロイするまで、パイプラインは以前のスキーマを使用してデータを読み取り続けます。

出力スキーマについては、新しい列を追加してもリプレイは必要ありません。ただし、出力スキーマから列を削除すると状態の断絶が発生し、リプレイが必要になります。

選択的なデータの再取り込み

出力ビューをリセットせずに、特定の時点からデータを再取り込みできます。再取り込みを選択すると、再取り込み時に出力に存在するすべてのデータが保持されるため、既存の出力の状態を維持しながら過去のデータを再処理できます。

この動作を設定するには、デプロイパネルのアドバンスドセクションを展開し、パイプラインをリプレイする際にリプレイ時に出力をリセットオプションを無効にします。

出力のインクリメンタル動作を強制する

Pipeline Builder のインクリメンタル実行を必須にする設定を使用して、パイプラインでインクリメンタル（差分処理）実行を強制できます。

この設定により、インクリメンタルに実行するよう設定されたジョブは、インクリメンタル実行ができない場合に自動的に失敗します。これにより、次のような意図しないスナップショット処理を防ぐことができます。

• 誤ってスナップショット化された入力
• 出力スキーマの変更によって強制されるスナップショット処理
• その他の予期しない全件更新

この機能は、PySpark のインクリメンタルトランスフォームでも、@incremental デコレーターに require_incremental=True を設定することで利用できます。

パイプラインでインクリメンタル実行を強制するには、以下の手順に沿って設定します。

1. Pipeline Builder でパイプラインを開きます。
2. デプロイボタンの右側または右側のツールバーにある Build 設定のスライダーアイコンを選択します。
3. 詳細な設定まで下にスクロールします。
4. インクリメンタル実行を必須にするを True に設定します。この設定はデフォルトでは無効です（値なしに設定されています）。

重要な考慮事項

• 有効にすると、ジョブのグループ化に関係なく、パイプライン内のすべてのインクリメンタル出力は、インクリメンタルに実行できない場合に失敗します。
• インクリメンタル実行を必須にする設定を True に設定できるのは、パイプラインにインクリメンタル入力または出力が1つ以上ある場合のみです。ない場合、このオプションを有効にするとデプロイエラーが発生します。
• インクリメンタル実行を必須とするよう設定されたパイプラインがスナップショットとして実行できるのは、パイプラインへの変更によって状態の断絶が必要になる場合のみです。実行前に、このことへの了承を求められます。

空のトランザクションをコミットする

Pipeline Builder のインクリメンタルパイプラインには、ビルドの出力行数が0行の場合に空のトランザクションをコミットできるビルド設定があります。有効にすると、Pipeline Builder は空のトランザクションをコミットし、後続のビルドで入力が再処理されるのを防ぎます。

この設定は、一部のビルドで出力行が生成されないことが正常な動作であるインクリメンタルパイプラインに役立ちます。この設定を有効にしない場合、出力を生成しなかった入力データは後続のビルドで再処理されます。

この設定を構成するには、次の手順を実行します。

1. Pipeline Builder でパイプラインを開きます。
2. デプロイボタンの右側または右側のツールバーにある Build 設定のスライダーアイコンを選択します。
3. 詳細な設定まで下にスクロールします。
4. 空のトランザクションをコミットオプションを有効にします。