トランザクションの中止

Python トランスフォームは、出力データセットが変更されない場合（データセットに新しいデータが書き込まれない場合）にジョブを正常に完了できるように、トランザクションの中止をサポートしています。これは、transform デコレーターを使用し、トランスフォーム内で利用可能な ctx オブジェクトで .abort_job() を呼び出すことで実現できます。

特定の条件下で出力データセットと下流のデータセットが更新されないようにする必要がある場合、トランザクションを中止できます。出力データセットが更新されると、下流のデータセットは最新ではない（古い）と見なされ、次回のビルド時（手動またはスケジュールされたビルド）に更新されます。トランザクションの中止は、ビルドを失敗させる代わりの手段となります。これにより、実際に問題が発生している場合を識別しやすくなります。

中止されたトランザクションは、データセットのトランザクション履歴で、グレー表示された成功したジョブとして表示されます。これにより、成功したビルドでトランザクションがコミットされたかどうかを一目で区別できます。

コミットされたトランザクションと中止されたトランザクションの比較。

トランザクションの中止が適している場合の例を以下に示します。

入力データセットの内容に基づいて、データセットを更新するカスタム条件がある場合。
データセットが古い状態にならないため、強制ビルドが必要な場合。たとえば、入力データセットではなく、API 呼び出しによってデータを取得するデータセットが該当します。
スケジュールに従って実行されると常に更新されるライトバックデータセットがある場合（詳細な例は後述します）。

常に更新されるものの、出力が必ずしも変更されるとは限らないデータセットに abort_job() を使用した検証を追加すると、下流での不要な更新を回避してコンピューティングリソースを節約できます。

例：カスタム条件の想定例

以下は、入力データセットに当日分のデータが到着した場合にのみ出力データセットが更新されるようにする、基本的な想定例です。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from transforms.api import transform, Input, Output
from datetime import date
import polars as pl

@transform.using(
    holiday_trips=Input('/examples/trips'),
    processed=Output('/examples/trips_processed')
)
def update_daily_trips(ctx, holiday_trips, processed):
    holiday_trips_df = holiday_trips.polars()
    todays_trips_df = holiday_trips_df.filter(
        pl.col("trip_date") == date.today()
    )
    if (todays_trips_df.height == 0):
      ctx.abort_job()
    else:
      processed.write_table(todays_trips_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output
from datetime import date

@transform.using(
    holiday_trips=Input('/examples/trips'),
    processed=Output('/examples/trips_processed')
)
def update_daily_trips(ctx, holiday_trips, processed):
    conn = ctx.duckdb().conn
    todays_trips_df = conn.sql("SELECT * FROM holiday_trips WHERE trip_date = DATE '{}'".format(date.today())).fetchdf()
    if (todays_trips_df.height == 0):
      ctx.abort_job()
    else:
      processed.write_table(todays_trips_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output
from datetime import date

@transform.using(
    holiday_trips=Input('/examples/trips'),
    processed=Output('/examples/trips_processed')
)
def update_daily_trips(ctx, holiday_trips, processed):
    holiday_trips_df = holiday_trips.pandas()
    todays_trips_df = holiday_trips_df[holiday_trips_df['trip_date'] == date.today()]
    if (todays_trips_df.shape[0] == 0):
      ctx.abort_job()
    else:
      processed.write_table(todays_trips_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output
from datetime import date

@transform.spark.using(
    holiday_trips=Input('/examples/trips'),
    processed=Output('/examples/trips_processed')
)
def update_daily_trips(ctx, holiday_trips, processed):
    holiday_trips_df = holiday_trips.dataframe()
    todays_trips_df = holiday_trips_df.filter(holiday_trips_df.trip_date == date.today())
    if (todays_trips_df.count() == 0):
      ctx.abort_job()
    else:
      processed.write_dataframe(todays_trips_df)

通常、1 行の確認はすべての行の件数を数えるより高速です

if (len(todays_trips_df.head(1)) == 0) を使用すると、すべての行の件数を不要に数えるのではなく、1 行以上存在するかどうかだけを確認するため、通常は if (todays_trips_df.count() == 0) よりも高速に結果が得られます。

中止されたトランザクションと無視（確認）されたジョブの違いは何ですか？

ジョブが「無視（確認）」として指定されると、Foundry が Job Spec は古くないと判断するため、計算は実行されません。トランザクションが中止された場合は、ジョブが実行され、正常に完了しますが、出力データセットは変更されず、トランザクションはコミットされません。

中止されたトランザクションと無視（確認）されたジョブの比較

中止されたトランザクションとインクリメンタル（差分処理）トランザクションにはどのような関係がありますか？

インクリメンタルトランスフォームは、コミットされたトランザクションのみを使用して、入力と出力の両方のデータセットビューを読み取ります。つまり、インクリメンタル計算を実行する際、中止されたトランザクションは無視（確認）されます。

インクリメンタルトランスフォームでジョブ全体のトランザクションを明示的に中止すると、次回のビルドでは、中止されたトランザクションが発生しなかったものとして入力を読み取り（したがって再処理し）、インクリメンタルに実行できます。

個別の出力を中止するとインクリメンタルビルドが機能しなくなります

PySpark を使用する場合、ジョブ全体ではなく、個別のデータセットで abort() を呼び出すことができます。ただし、この方法はインクリメンタル計算で問題を引き起こす可能性があるため、推奨されません。

特に、出力の一部でのみトランザクションを中止すると、ビルドをインクリメンタルに実行できなくなります。トランザクションが中止された出力では、中止されたトランザクションが無視（確認）されるため、出力の Job Spec は以前の入力トランザクション範囲を使用します。トランザクションがコミットされた出力では、出力の Job Spec は現在の入力トランザクション範囲を使用します。この入力トランザクション範囲の不一致により、トランスフォームをインクリメンタルに実行できなくなります。

複数の出力を持つインクリメンタルトランスフォームで、出力の一部のトランザクションを明示的に中止すると、次回のビルドではインクリメンタル計算のチェックに失敗して Provenance records for the previous build are inconsistent が表示され、スナップショットとして実行されます。require_incremental=True が設定されている場合、ビルドは InconsistentProvenanceRecords エラーで失敗します。これは、出力の現在のビューが異なる入力トランザクションによって生成されているためです。

インクリメンタルビルドを実行する場合は、個別の出力を中止するのではなく、TransformContext の .abort_job() メソッドを使用してジョブ全体を中止してください。これにより、ビルドの_すべて_の出力が中止されます。v2_semantics を使用している場合、後続のビルドをインクリメンタルに実行できる状態を維持しながら出力を中止するには、この方法しかありません。

←

PREVIOUS使用ガイド

NEXTスナップショットからの履歴データセットの作成

→