Pipeline Builder でインクリメンタル（差分処理）パイプラインを作成する

このチュートリアルでは、Pipeline Builder を使用して、1つのデータセットを出力するシンプルなインクリメンタルパイプラインを作成します。

以下で使用するデータセットは、インクリメンタル計算を適用できるケースを示すための架空の例です。

Python で変換ロジックを記述するには、Pipeline Builder で Python ユーザー定義関数を使用するか、Python コードでインクリメンタルトランスフォームを作成できます。

パート1：課題の説明

毎週新しいデータが追加される入力データセット flights があるとします。このデータを JFK 空港から出発するフライトだけに絞り込み、それらのフライトだけを出力 filtered_flights に追加したいと考えています。

flights データセットには2000万行ありますが、毎週追加されるのは100万行だけだとします。インクリメンタル計算では、スナップショット計算のようにすべての行を処理するのではなく、flights の最新の未処理トランザクションだけを処理すれば済みます。

以降のセクションでは、インクリメンタルパイプラインを設定する手順を説明します。

パート2：インクリメンタル処理の要件を確認する

まず、インクリメンタル処理に関するすべての制約を満たしていることを確認します。

• 入力 flights が、APPEND トランザクション、または既存のファイルを変更しない UPDATE トランザクションによって更新されていること。
• flights から filtered_flights を計算するロジックが、後続のビルドで、すでに filtered_flights に書き込まれているデータの変更を必要としないこと。
  • パイプラインのロジックを変更したい場合（たとえば、LGA 空港から出発するフライトも含める場合）、パイプラインを更新できます。そのロジックを処理済みのフライトにも適用したい場合は、パイプラインのリプレイを検討してください。
• パイプラインにウィンドウ関数、集計、またはピボットが含まれる場合、それらが現在のトランザクションのみを対象として動作することを意図していることを確認してください。

考慮事項の一覧については、Pipeline Builder のインクリメンタル計算に関する重要な制限を参照してください。

パート3：パイプラインを作成する

ここで、新しいパイプラインを初期化できます（ステップごとのウォークスルーについては、Pipeline Builder でバッチパイプラインを作成するを参照してください）。flights を入力データセットとしてインポート済みとします。

まず、データセットの下にあるボタンを使用して、入力データセットをインクルメンタル（差分処理） に指定します。変更を示す青いバッジが右上隅に表示されます。

次に、flights を JFK 空港から出発するフライトに絞り込むデータ加工を追加します。データセット入力の右側にある、ツールチップにインクリメンタル入力と表示されるアイコンに注目してください。下流のデータ加工には、インクリメンタルに処理されることを示すこのアイコンが表示されます。

グラフ上では、下流ノードに入力と同じ青いバッジが表示されます。

最後に、出力データセット filtered_flights を追加します。

インクリメンタルパイプラインでは、出力にインクリメンタル動作を強制し、インクリメンタルに実行できない場合はビルドを失敗させるオプションを利用できます。

パート4：出力データセットをデプロイする

これで、パイプラインをデプロイする準備が整いました。

デプロイ時にリプレイする

過去の入力トランザクションを再処理する必要が生じることがあります（たとえば、ロジックが変更され、以前のバージョンの出力データが古くなった場合）。このような場合は、デプロイ時にリプレイを選択して、入力全体をパイプラインのロジックで処理できます。リプレイ後は、新しい APPEND トランザクションが入力に追加されるたびに、パイプラインはインクリメンタル計算を継続するはずです。