インクリメンタル入力のバッチサイズを制限する

通常、出力データセットがインクリメンタル（差分処理）にビルドされると、各入力データセットの未処理のトランザクションがすべて同じジョブで処理されます。ただし、状況によっては、ジョブで処理されるトランザクション数が大きく変わることがあります。

• インクリメンタルトランスフォームが SNAPSHOT モードでビルドされ、入力全体が最初から読み取られる場合（たとえば、トランスフォームのセマンティックバージョンが引き上げられた場合）。
• 入力データセットにデータのトランザクションが多数蓄積され、下流のインクリメンタルトランスフォームで1つのジョブで多数のトランザクションを処理する必要がある場合。

トランスフォームの各インクリメンタル入力にトランザクション上限を設定して、各ジョブで読み取るデータ量を制限できます。

以下の例では、次の入力を使用するようにインクリメンタルトランスフォームを設定します。

• それぞれ異なるトランザクション上限を持つ2つのインクリメンタル入力
• トランザクション上限を使用しないインクリメンタル入力
• スナップショット入力

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from transforms.api import transform, Input, Output, incremental

@incremental(
    v2_semantics=True,
    strict_append=True,
    snapshot_inputs=["snapshot_input"]
)
@transform(
    # 最大3件のトランザクションを読み取るように設定されたインクリメンタル（差分処理）入力
    input_1=Input("/examples/input_1", transaction_limit=3),

    # 最大2件のトランザクションを読み取るように設定されたインクリメンタル入力
    input_2=Input("/examples/input_2", transaction_limit=2),

    # トランザクション数の制限がないインクリメンタル入力
    input_3=Input("/examples/input_3"),

    # 毎回ビュー全体が読み取られるスナップショット入力
    snapshot_input=Input("/examples/input_4"),

    output=Output("/examples/output")
)
def compute(input_1, input_2, input_3, snapshot_input, output):
    ...

出力を最新に保つビルドスケジュールを作成する

トランザクション上限を有効にすると、データの一部しか処理されないため、ビルドが成功しても、データセットが上流の最新データに追いついていない場合があります。以下の手順でスケジュールを設定すると、出力データセットが入力に対して最新になるまでビルドを続けられます。

1. Data Lineage に移動します。
2. 右側のパネルでスケジュールの管理タブを開き、新規スケジュールを作成 を選択します。
3. 対象リソースを、トランザクション上限が設定されたインクリメンタルトランスフォームの出力データセットに設定します。
4. Build のタイミングセクションで選択肢を選び、必要に応じて頻度の詳細を設定します。

• 特定のリソースが更新されたとき
• 特定の時刻
• 手動でトリガーされたとき
• 複数の時間条件またはイベント条件が満たされたとき

5. パネルの下部までスクロールし、詳細設定セクションを展開します。
6. Build の正常完了時にスケジュールを再トリガーするオプションを有効にします。
7. 画面右上の保存を選択して、スケジュール設定を保存します。

ジョブのトランザクション範囲を表示する

以下の手順で、インクリメンタルジョブで入力ごとに読み取られたトランザクションの範囲を確認できます。

1. 調べたいジョブの Spark の詳細ページに移動します。
2. スナップショット/インクリメンタルタブを選択します。

このページでは、入力ごとにトランザクションの範囲が報告され、今回と前回のジョブで各入力のどの部分が処理されたかが表示されます。

• 現在のビューの範囲： 今回のジョブで読み取られた入力のビューの開始トランザクションと終了トランザクションを示します。
• 処理バッチの範囲：「現在のビューの範囲」のうち、今回のジョブで処理されたバッチの開始トランザクションと終了トランザクションを示します。
• 前回の終了トランザクション： 前回のジョブにおける入力のビューの最後のトランザクションを示します。
• 最終読み取りトランザクション： 前回のジョブで処理された入力の最後のトランザクションを示します。このトランザクションは、次のいずれかになります。
  • 次のいずれかに該当する場合は、「前回の終了トランザクション」と同じになります。
    • 前回のジョブで、入力がトランザクション上限なしで処理された場合。
    • 前回のジョブで、入力にトランザクション上限を使用する設定がされており、処理されたバッチが入力の最後のバッチだった場合。
  • 「前回の終了トランザクション」より前のトランザクションになります。これは、入力でトランザクション上限が使用され、処理されたバッチが入力の最後のバッチでは_なかった_場合に発生します。

トランザクションを選択すると、入力の履歴ページに移動し、該当するトランザクションが強調表示されます。

トランザクション上限のある入力の読み取り範囲を理解する

入力にトランザクション上限が設定されている場合も、設定されていない場合も、同じ added、current、previous の読み取り範囲を使用できますが、動作は少し異なります。

以下の例では、トランザクション A から C までをすでに処理したインクリメンタルトランスフォームを考えます。ここで、比較的多数のトランザクション D から J までが入力に追加されたとします。

トランザクション上限なしで入力を読み取る場合、次のジョブでの読み取りモードごとのトランザクション範囲は次のようになります。

• 追加分：D から J
• 前回：A から C
• 現在：A から J

一方、トランザクション上限を3に設定して入力を読み取る場合は、入力に追いつくまでに_3回_のジョブが必要です。各ジョブでの読み取りモードごとのトランザクション範囲は次のようになります。

1回目のジョブ：

• 追加分：D から F（次の3件の未処理トランザクション）
• 前回：A から C（前回のジョブで処理されたすべてのトランザクション）
• 現在：A から F（このバッチまでに処理された、このバッチを含むすべてのトランザクション）

2回目のジョブ：

• 追加分：G から I
• 前回：A から F
• 現在：A から I

3回目のジョブ：

• 追加分：J
• 前回：A から I
• 現在：A から J

ここで、出力がスナップショットされた場合（たとえば、セマンティックバージョンが変更された場合）、入力の開始トランザクションから再び処理され、読み取り範囲は以下のようになります。

トランザクション上限なしの場合：

• 追加分：A から J（すべてのトランザクションが「未処理」になります）
• 前回： なし
• 現在：A から J

インクリメンタル入力の場合：

1回目のジョブ：

• 追加分：A から C
• 前回： なし
• 現在：A から C（このバッチまでに処理された、このバッチを含むすべてのトランザクション）

2回目のジョブ：

• 追加分：D から F
• 前回：A から C
• 現在：A から F

3回目のジョブ：

• 追加分：G から I
• 前回：A から F
• 現在：A から I

4回目のジョブ：

• 追加分：J
• 前回：A から I
• 現在：A から J

要件と制限

インクリメンタル（差分処理）トランスフォームでトランザクション制限を使用するには、必要なツールとサービスにアクセスできること、およびトランスフォームとデータセットが以下の要件を満たしていることを確認してください。

トランスフォームは、次の条件を満たす必要があります。

• インクリメンタルデコレーターを使用し、v2_semantics 引数を True に設定していること。
• Python トランスフォームのバージョン 3.25.0 以降を使用するように設定されていること。特定のバージョンの Python トランスフォームを使用するには、モジュールの固定でジョブを設定します。
• 軽量トランスフォームではないこと。

入力データセットにトランザクション制限を設定するには、次の条件を満たす必要があります。

• トランザクション型データセットの入力であること。
• データセットの現在のビューに、APPEND トランザクションのみが含まれていること。ただし、最初のトランザクションは SNAPSHOT でもかまいません。
• スナップショット入力ではないこと。