インクリメンタル（差分処理）の使用ガイド

インクリメンタルデコレーター

incremental() デコレーターを使用すると、インクリメンタル計算を有効にするロジックでトランスフォームの計算関数をラップできます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
import polars as pl
from transforms.api import transform, incremental, Input, Output

@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.polars(mode='added')  # 'added' がデフォルトなので、students.polars() だけでも構いません
    processed.write_table(students_df.filter(pl.col("hair") == 'Brown'))

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

@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(ctx, students, processed):
    conn = ctx.duckdb().conn
    # DuckDB はインクリメンタル（差分処理）読み取りモードに自動的に対応します
    filtered_query = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")
    processed.write_table(filtered_query)

Copied!
1
2
3
4
5
6
7
8
@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.pandas(mode='added')  # 'added' がデフォルトなので、students.pandas() だけでも構いません
    processed.write_table(students_df[students_df.hair == 'Brown'])

Copied!
1
2
3
4
5
6
7
8
9
10
from transforms.api import transform, incremental, Input, Output

@incremental()
@transform.spark.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.dataframe('added')
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))

incremental() デコレーターは、transform()、transform.using()、transform_df()、または transform_pandas() デコレーターを使用する既存のトランスフォームをラップするために使用できます。トランスフォームの計算関数は、インクリメンタルと非インクリメンタルの両方での実行に対応している必要があります。incremental() デコレーターは、主に次の2つの処理を行います。

• トランスフォームが前回のビルドに関する情報を参照できるようにします。incremental() デコレーターは、この情報を使用して、以下の要件に従ってトランスフォームをインクリメンタルに実行できるかどうかを判断します。
• 入力、出力、およびコンテキストのオブジェクトを、追加の機能を提供するインクリメンタルサブクラスに変換します。具体的には、TransformInput は IncrementalTransformInput に、TransformOutput は IncrementalTransformOutput に、TransformContext は IncrementalTransformContext になります。これらのインクリメンタルオブジェクトは、デコレーターでラップされたトランスフォームに渡されます。

インクリメンタルデコレーターは7個の引数を取ります。

Copied!
1
2
3
4
5
6
7
8
9
transforms.api.incremental(
    require_incremental=False,
    semantic_version=1,
    snapshot_inputs=None,
    allow_retention=False,
    allow_user_deletes=False,
    strict_append=False,
    v2_semantics=False
)

require_incremental 引数を True に設定すると、トランスフォームをインクリメンタルに実行できない場合に失敗します。require_incremental=True であっても、次の2つの場合はトランスフォームをスナップショットとして実行できます。

1. 出力のいずれかが、これまで一度もビルドされていない場合。
2. セマンティックバージョンが変更され、スナップショットが明示的に要求された場合。

トランスフォームをインクリメンタルに実行できなかった原因をデバッグするには、ドライバーログで警告 transforms.api._incremental: Not running incrementally を探します。

:func:~transforms.api.incremental デコレーターの semantic_version 引数を使用すると、トランスフォームの次回の実行を強制的に非インクリメンタルにできます。

• 今回の実行のセマンティックバージョンが前回の実行のセマンティックバージョンと異なる場合、トランスフォームは非インクリメンタルに実行されます。
• 値を指定しない場合は、セマンティックバージョンは 1 に設定されます。
• 前回の実行のセマンティックバージョンが存在しない場合（たとえば、既存のトランスフォームをインクリメンタルトランスフォームに変換する場合）、値は 1 とみなされます。これにより、新しいスナップショットを必要とせずに、トランスフォームのインクリメンタル実行を開始できます。
• トランスフォームの後続の実行を強制的に非インクリメンタルにするには、@incremental() デコレーターの semantic_version 引数の値を上げます。

snapshot_inputs 引数を使用すると、一部の入力をスナップショット入力として定義できます。スナップショット入力は、スナップショット以外の入力とは異なり、更新および削除による変更に対応しています。詳細については、スナップショット入力を参照してください。

allow_retention 引数を True に設定すると、トランスフォームのインクリメンタル性を維持しながら、保持ポリシーによって入力データセットと出力データセット内のファイルを削除できるようになります。

allow_user_deletes 引数を True に設定すると、トランスフォームのインクリメンタル性を維持しながら、保持ポリシーによらず入力データセット内のファイルを削除できるようになります。この引数には v2_semantics=True が必要であり、軽量トランスフォームでは利用できません。

strict_append パラメーターが True に設定され、入力データセットがインクリメンタルである場合、基になる Foundry のトランザクションの種類は APPEND に設定され、インクリメンタル書き込みには APPEND トランザクションが使用されます。既存のファイルを上書きしようとすると、例外が発生します。 入力データセットがインクリメンタルでない場合、strict_append は SNAPSHOT として実行されます。コードが APPEND としてインクリメンタルに実行されるようにするには、require_incremental=True を使用してください。既存のファイルを上書きしようとすると、上書きに成功します。 書き込み操作では、Parquet の概要メタデータや Hadoop の SUCCESS ファイルなどの補助ファイルも含め、どのファイルも上書きできません。すべての Foundry 形式のインクリメンタル書き込みは、strict_append モードに対応しているはずです。

v2_semantics パラメーターが True に設定されている場合、v2 インクリメンタルセマンティクスが使用されます。v2 と v1 のインクリメンタルセマンティクスで動作に違いはないはずであり、すべてのユーザーに、このパラメーターを True に設定することを推奨します。Catalog 以外の入力リソースと出力リソースをインクリメンタルに読み書きできるのは、v2 セマンティクスを使用する場合のみです。

重要な情報

前述のとおり、incremental() デコレーターでラップしたトランスフォームの計算関数は、インクリメンタルと非インクリメンタルの両方での実行に対応している必要があります。デフォルトの読み取りモードと書き込みモード（このガイドの以降の部分で詳しく説明します）は、この2つのロジックへの対応に役立ちますが、計算コンテキストの is_incremental プロパティに基づいて分岐する必要がある場合もあります。

もう1つの重要な点は、incremental() デコレーターを transform_df() または transform_pandas() とともに使用する場合、デフォルトの読み取りモードと書き込みモードしか使用できないことです。追加される出力行が、追加された入力行のみから算出されるトランスフォームであれば、これで十分です（追加の例を参照してください）。ただし、結合、集計、重複排除など、入力の読み取りモードや出力の書き込みモードの設定が必要となる、より複雑なロジックをトランスフォームで実行する場合は、incremental() デコレーターを transform() とともに使用してください。インクリメンタルデコレーターを transform() とともに使用すると、読み取りモードと書き込みモードを設定できます。

入力と出力のインクリメンタル（差分処理）モード

インクリメンタルトランスフォームは強力な機能です。ただし、初めて使用する場合、API は直感的に理解しにくいことがあります。インクリメンタルトランスフォームは、VS Code プレビューでもビルド時と同じように動作します。API を試して実際の動作を理解する際には、VS Code プレビューの使用を推奨します。

構造化データセットと非構造化データセットの両方を扱えるように、dataframe() メソッドと filesystem() メソッドでは同じ読み取りモードを使用します。

IncrementalTransformInput

transforms.api.IncrementalTransformInput オブジェクトは、データ読み取りメソッドを拡張して、任意の読み取りモードパラメーターを受け取れるようにします。

インクリメンタルデコレーターを使用してトランスフォームを定義した場合、トランスフォームをインクリメンタルに実行するか、非インクリメンタルに実行するかによって、読み取りモードの動作が異なります。

デフォルトの読み取りモードは added です。各読み取りモードが返す内容を示すコードについては、インクリメンタルトランスフォームの例を参照してください。

トランスフォームに incremental() が指定されていても、入力をインクリメンタルに扱うことが望ましくない場合があります。詳細およびこのような入力で読み取りモードの動作がどのように異なるかについては、スナップショット入力セクションを参照してください。

出力のデフォルトの読み取りモードは current で、利用可能な出力読み取りモードは added、current、previous です。出力読み取りモードの詳細については、以下のセクションを参照してください。

インクリメンタルトランスフォームの性質上、入力ビューを構築するために、入力データセットの最後の SNAPSHOT トランザクション以降のすべての過去のトランザクションを読み込みます。インクリメンタルトランスフォームが徐々に遅くなる場合は、インクリメンタル入力データセットで SNAPSHOT ビルドを実行することを推奨します。

IncrementalTransformOutput

transforms.api.IncrementalTransformOutput オブジェクトを使用すると、出力データセットの読み取りモードと書き込みモードにアクセスできます。インクリメンタル（差分処理）ビルドと非インクリメンタルビルドの両方に対応するロジックを書くうえで重要なのは、デフォルトの書き込みモードである modify です。書き込みモードには2種類あります。

• modify：このモードでは、ビルド中に書き込まれたデータで既存の出力を変更します。たとえば、出力が modify モードのときに write_dataframe() または write_table() を呼び出すと、書き込まれた DataFrame が既存の出力に追加されます。
• replace：このモードでは、ビルド中に書き込まれたデータで出力を完全に置き換えます。

トランスフォームがインクリメンタルに実行される場合、出力のデフォルトの書き込みモードは modify に設定されます。トランスフォームが非インクリメンタルに実行される場合、出力のデフォルトの書き込みモードは replace に設定されます。

入力 DataFrame のデフォルトの読み取りモードは added であることを思い出してください。入力のデフォルトの読み取りモードが added で、出力のデフォルトの書き込みモードが modify であるため、インクリメンタルビルドと非インクリメンタルビルドの両方に対応するロジックをより簡単に書けます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import polars as pl

@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def incremental_filter(students, processed):
    # これまで読み取っていない行だけを読み取ります。
    new_students_df = students.polars()  # これは students.polars('added') と同等です

    # 非インクリメンタルの場合は、すべての行を読み取り、出力を置き換えます。
    # インクリメンタルの場合は、新しい行だけを読み取り、出力に追加します。
    processed.write_table(
        new_students_df.filter(pl.col("hair") == 'Brown')
    )

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def incremental_filter(ctx, students, processed):
    # これまで読み取っていない行だけを読み取ります。
    conn = ctx.duckdb().conn
    new_students_df_filtered = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")

    # 非インクリメンタルの場合は、すべての行を読み取り、出力を置き換えます。
    # インクリメンタルの場合は、新しい行だけを読み取り、出力に追加します。
    processed.write_table(
        new_students_df_filtered
    )

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def incremental_filter(students, processed):
    # これまで読み取っていない行だけを読み取ります。
    new_students_df = students.pandas()  # これは students.pandas('added') と同等です

    # 非インクリメンタルの場合は、すべての行を読み取り、出力を置き換えます。
    # インクリメンタルの場合は、新しい行だけを読み取り、出力に追加します。
    processed.write_table(
        new_students_df[new_students_df.hair == 'Brown']
    )

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@incremental()
@transform.spark.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def incremental_filter(students, processed):
    # これまで読み取っていない行だけを読み取ります。
    new_students_df = students.dataframe()  # これは students.dataframe('added') と同等です

    # 非インクリメンタルの場合は、すべての行を読み取り、出力を置き換えます。
    # インクリメンタルの場合は、新しい行だけを読み取り、出力に追加します。
    processed.write_dataframe(
        new_students_df.filter(new_students_df.hair == 'Brown')
    )

インクリメンタル計算のより複雑なユースケースでは、正しい書き込みモードを算出し、手動で設定する必要がある場合があります。これは、インクリメンタル出力の set_mode() メソッドを使用して行うことができ、transform() または transform.using() デコレーターを使用する場合にサポートされます。これらのデコレーターでは、出力を保存するための書き込みメソッドを明示的に呼び出す前に、set_mode() を呼び出せます。一方、transform_df() および transform_pandas() デコレーターは、DataFrame の出力を保存するために write_dataframe() および write_pandas() を呼び出すため、これらのデコレーターで使用される書き込みモードは incremental() デコレーターによって決まります。

書き込みモードに加えて、transforms.api.IncrementalTransformOutput を使用すると、出力データセットから DataFrame を読み取ることもできます。これには、任意の読み取りモードパラメーターを受け取る pandas()、polars()、dataframe() などのデータ読み取りメソッドを使用します。transform.using() を使用する場合は、モードをキーワード引数として指定します：output.pandas(mode='previous')。デフォルトの読み取りモードは current に設定されており、ほかに利用可能な出力の読み取りモードは added と previous です。読み取りモードの動作は、データセットの書き込みモードの設定によって異なります。

前回の実行のデータを読み取るための有効な組み合わせ

トランスフォームの前回の出力にアクセスするには、トランスフォームがインクリメンタルに実行されている必要があります（ctx.is_incremental is True）。トランスフォームが非インクリメンタルに実行されている場合、通常は前回の出力にアクセスできる読み取りモードでも、前回の出力データの行は返されません。以下のテーブルに示す動作は、インクリメンタル処理の意味と、current および previous 読み取りモードのトランザクションがトランスフォームの実行開始時に確定されるという事実に基づいています。

前回の DataFrame を読み取る際に指定するスキーマ ↗は、トランスフォームが非インクリメンタルに実行されている場合に、空の DataFrame を生成するために使用されます。トランスフォームがインクリメンタルに実行されている場合、このスキーマは前回の出力の実際のスキーマと比較されます。列のデータ型、列の null 許容性、または列の順序が一致しない場合、例外が発生します。列の順序を同じに保つには、次の構文を使用できます。

Copied!
1
2
3
previous = out.dataframe('previous', schema)  # schema は pyspark.sql.types.StructType です

out.write_dataframe(df.select(schema.fieldNames()))

詳しくは、マージと置換の例を参照してください。

トランスフォームの前回の呼び出しで書き込まれたデータにアクセスするための組み込みの方法はありません。そのデータが必要な場合は、補助的な出力データセットを追加し、トランスフォームの実行ごとに、書き込まれた出力のコピー（または関連するメタデータ）を replace 書き込みモードでそこに書き込めます。これにより、今後、前回のトランスフォームで書き込まれた出力だけを参照できるようになります。

現在の実行で書き込まれたデータを読み取るための有効な組み合わせ

意図がより明確になるため、added を推奨します。 現在のトランスフォームで書き込まれたデータの読み取りが役立つのは、データセットの内容全体を検証し、検証に失敗した場合にビルドを失敗させるような場合です。この方法では、チェックを実行するために Spark でデータを再計算したりキャッシュしたりする必要がありません。

IncrementalTransformContext

TransformContext オブジェクトと比較すると、IncrementalTransformContext オブジェクトには追加のプロパティ is_incremental があります。このプロパティは、トランスフォームがインクリメンタル（差分処理）に実行される場合に True に設定されます。これは次のことを意味します。

• 出力のデフォルトの書き込みモードは modify に設定されます。
• 入力のデフォルトの読み取りモードは added に設定されます。

インクリメンタルモードの概要

インクリメンタルデコレーターでは、読み取りモードに「previous」を指定することで、トランスフォームの以前の入力と出力にアクセスできます。これにより、過去のコンテキストに基づいて現在のビルドを実行できます。トランスフォームがスナップショットモードで実行される場合、「previous」のデータフレームは空になります。これは、初回の実行であるか、ロジックやデータが大幅に変更されて再計算が必要になったためです。

ただし、最も一般的なのは、入力に「added」モード、出力に「modify」モードを使用するケースです。これらのモードはデフォルトで使用されます。これにより、入力データセットから新しく追加された行を取得して処理し、出力データセットに追加できます。

出力に行を追加する代わりに、出力データセットにすでに存在する行を変更したい場合もあります。その場合は、一般的なシナリオの例で示されているように、「replace」モードを使用します。

インクリメンタル（差分処理）計算の要件

以下では、髪の色が茶色の学生だけに絞り込むインクリメンタルトランスフォームを分析します。

Copied!
1
2
3
4
5
6
7
8
9
10
import polars as pl

@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.polars()
    processed.write_table(students_df.filter(pl.col("hair") == 'Brown'))

Copied!
1
2
3
4
5
6
7
8
9
@incremental()
@transform.using(
students=Input('/examples/students_hair_eye_color'),
processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(ctx, students, processed):
    conn = ctx.duckdb().conn
    filtered_query = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")
    processed.write_table(filtered_query)

Copied!
1
2
3
4
5
6
7
8
@incremental()
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.pandas()
    processed.write_table(students_df[students_df.hair == 'Brown'])

Copied!
1
2
3
4
5
6
7
8
@incremental()
@transform.spark.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.dataframe()
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))

入力データセット /examples/students_hair_eye_color が、新しい学生のデータで完全に置き換えられたとします。この場合、新しい学生のデータを以前の出力に追加すると、出力データセットは不正確になります。これは、incremental() デコレーターがトランスフォームをインクリメンタルに実行しないと判断する状況の例です。

トランスフォームをインクリメンタルに実行するには、次の要件を満たす必要があります。

• トランスフォームのスナップショット入力以外のすべての入力データセットで、前回の実行以降にファイルの追加のみが行われた、保持ポリシーによる削除が行われた、または allow_user_deletes で許可されたユーザーによる削除が行われたこと。
• 入力の最後のトランザクションが、既存のファイルを変更する UPDATE、または allow_retention や allow_user_deletes で許可されていないファイルを削除する DELETE だった場合、その入力をインクリメンタルに処理できるのは、スナップショット入力として指定されている場合のみです。
• トランスフォームの入力のリストは、ビルドのインクリメンタル性を損なわずに変更できます。
• 複数の出力を持つトランスフォームでは、すべての出力が前回同じトランスフォームでビルドされていること。

トランスフォームに incremental() デコレーターが指定されていても、上記の要件のいずれかが満たされていない場合、トランスフォームは自動的に非インクリメンタルで実行されます。つまり、デフォルトの出力書き込みモードは modify ではなく replace に設定され、入力は非インクリメンタルで提供されます。また、以前の履歴にアクセスできないため、トランスフォーム内で出力から読み取ると、空のデータフレームが返されます。同様に、入力も非インクリメンタルで提供されます。

トランスフォームをインクリメンタルでのみ実行するように強制するには、incremental() デコレーターに require_incremental=True 引数を渡します。トランスフォームをインクリメンタルに実行できない場合、非インクリメンタルでの実行を試みるのではなく、意図的に失敗します。ただし、そのトランスフォームがこれまで一度も実行されていない場合、またはセマンティックバージョンが引き上げられた場合は、スナップショットとして実行することが引き続き許可されます。

特定の入力を完全に書き換えても、トランスフォームをインクリメンタルに実行できるようにしておきたい場合がよくあります。そのためには、それらの入力をスナップショット入力として設定します。

追加のみの入力変更

前回の実行以降、すべてのインクリメンタル入力に対してファイルの追加のみが行われている場合（APPEND または UPDATE トランザクションによる追加）、トランスフォームをインクリメンタルに実行できます。

逆に、インクリメンタル入力について次のいずれかが当てはまる場合、トランスフォームをインクリメンタルに実行できません。

• インクリメンタル入力が完全に書き換えられた場合（たとえば、SNAPSHOT トランザクションが行われた場合）。
• インクリメンタル入力に対して、新しいファイルの追加だけではなく既存のファイルを変更または上書きする UPDATE トランザクションが行われた場合、または allow_retention や allow_user_deletes で許可されていない DELETE トランザクションが行われた場合。

たとえば、students_hair_eye_color 内の学生のリストが完全に変わった場合、以前の絞り込み済みの学生の出力は無効になり、置き換える必要があります。

保持ポリシーによる削除を含む入力

上流データセットが際限なく増加する場合に、下流の計算のインクリメンタル性に影響を与えずに古い行を削除できるようにするには（Foundry の保持ポリシーを使用）、そのデータセットに依存するインクリメンタルトランスフォームで、保持ポリシーが適用された入力を許可するように明示的に設定する必要があります。これは、transforms.api.incremental デコレーターの allow_retention 引数を使用して設定できます。

• このフィールドが True に設定されている場合、入力がインクリメンタル性を維持しているかどうかを評価する際に、保持ポリシーによるすべての削除が無視されます。つまり、保持ポリシーによって入力データセットから removed になったファイルはインクリメンタル性を損なわず、入力データセットの added 以外のファイル変更が保持ポリシーによる削除のみである場合、トランスフォームは引き続きインクリメンタルに実行されます。
• このフィールドが False（デフォルト）の場合、入力データセットに removed 型の変更があると、トランスフォームはスナップショットを実行します。

Copied!
1
2
3
4
5
6
7
8
9
10
import polars as pl

@incremental(allow_retention=True)
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.polars()
    processed.write_table(students_df.filter(pl.col("hair") == 'Brown'))

Copied!
1
2
3
4
5
6
7
8
9
@incremental(allow_retention=True)
@transform.using(
students=Input('/examples/students_hair_eye_color'),
processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(ctx, students, processed):
    conn = ctx.duckdb().conn
    filtered_query = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")
    processed.write_table(filtered_query)

Copied!
1
2
3
4
5
6
7
8
@incremental(allow_retention=True)
@transform.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.pandas()
    processed.write_table(students_df[students_df.hair == 'Brown'])

Copied!
1
2
3
4
5
6
7
8
@incremental(allow_retention=True)
@transform.spark.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.dataframe()
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))

上記の例では、データセット /examples/students_hair_eye_color に対して added の変更と保持ポリシーによる removed の変更のみが行われた後にトランスフォームを実行すると、インクリメンタルに実行されます。ほかの方法による removed の変更、または modified の変更がある場合、スナップショットが実行されます。

ユーザーによる削除を含む入力

依存する計算のインクリメンタル性に影響を与えずに、Foundry の保持ポリシー以外で古い行を削除するには、そのデータセットに依存するインクリメンタルトランスフォームで、ユーザーによる削除を許可するように明示的に設定する必要があります。これは、transforms.api.incremental デコレーターの allow_user_deletes 引数を使用して設定できます。

• このフィールドが True に設定されている場合、入力がインクリメンタル性を維持しているかどうかを評価する際に、保持ポリシー以外によるすべての削除が無視されます。つまり、保持ポリシー以外の方法で入力データセットから removed になったファイルはインクリメンタル性を損なわず、入力データセットの added 以外のファイル変更がユーザーによる削除のみである場合、トランスフォームは引き続きインクリメンタルに実行されます。
• このフィールドが False（デフォルト）の場合、入力データセットに保持ポリシー以外による removed 型の変更があると、トランスフォームはスナップショットを実行します。

Copied!
1
2
3
4
5
6
7
8
@incremental(v2_semantics=True, allow_user_deletes=True)
@transform.spark.using(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_hair_color(students, processed):
    students_df = students.dataframe()
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))

上記の例では、データセット /examples/students_hair_eye_color に対して added の変更と保持ポリシー以外による removed の変更のみが行われた後にトランスフォームを実行すると、インクリメンタルに実行されます。保持ポリシーに関連する removed の変更、または modified の変更がある場合、スナップショットが実行されます。保持ポリシーによる削除とユーザーによる削除の両方を許容するには、allow_retention=True と allow_user_deletes=True を両方設定します。

allow_user_deletes 引数には v2_semantics=True が必要であり、軽量トランスフォームでは利用できません。

スナップショット入力

入力が完全に書き換えられても、トランスフォームのインクリメンタル性が無効にならない場合があります。たとえば、電話番号の国番号を国に対応付け、定期的に書き換えられる単純な参照データセットがあるとします。このデータセットへの変更は、過去の計算結果を必ずしも無効にするわけではないため、トランスフォームのインクリメンタル実行を妨げるべきではありません。

前述のとおり、デフォルトでは、トランスフォームの前回の実行以降に入力のいずれかが完全に書き換えられた場合、トランスフォームをインクリメンタルに実行できません。スナップショット入力はこのチェックの対象外であり、開始トランザクションが実行ごとに異なっていても許容されます。

スナップショット入力は、incremental() デコレーターの snapshot_inputs 引数を使用して設定できます。

インクリメンタル（差分処理）トランスフォームで Cipher リソースと bellaso_python_lib を使用する場合、これらの暗号化器、復号器、およびハッシャーをスナップショット入力として指定する必要があります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@incremental(snapshot_inputs=['country_codes'])
@transform.using(
    phone_numbers=Input('/examples/phone_numbers'),
    country_codes=Input('/examples/country_codes'),
    output=Output('/examples/phone_numbers_to_country')
)
def map_phone_number_to_country(phone_numbers, country_codes, output):
    # これは前回の実行以降に追加された未読み取りの電話番号すべてになります
    phone_numbers_df = phone_numbers.polars()
    # これは以前に読み取った内容に関係なく、すべての国コードになります
    country_codes_df = country_codes.polars()

    output.write_table(
        phone_numbers_df.join(country_codes_df, left_on='country_code', right_on='code', how='left')
    )

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@incremental(snapshot_inputs=['country_codes'])
@transform.using(
    phone_numbers=Input('/examples/phone_numbers'),
    country_codes=Input('/examples/country_codes'),
    output=Output('/examples/phone_numbers_to_country')
)
def map_phone_number_to_country(ctx, phone_numbers, country_codes, output):
    conn = ctx.duckdb().conn
    joined_query = conn.sql("""
        SELECT *
        FROM phone_numbers
        LEFT JOIN country_codes
        ON phone_numbers.country_code = country_codes.code
    """)
    output.write_table(joined_query)

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@incremental(snapshot_inputs=['country_codes'])
@transform.using(
    phone_numbers=Input('/examples/phone_numbers'),
    country_codes=Input('/examples/country_codes'),
    output=Output('/examples/phone_numbers_to_country')
)
def map_phone_number_to_country(phone_numbers, country_codes, output):
    # これは前回の実行以降に追加された未読み取りの電話番号すべてになります
    phone_numbers_df = phone_numbers.pandas()
    # これは以前に読み取った内容に関係なく、すべての国コードになります
    country_codes_df = country_codes.pandas()

    output.write_table(
        phone_numbers_df.merge(country_codes_df, left_on='country_code', right_on='code', how='left')
    )

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@incremental(snapshot_inputs=['country_codes'])
@transform.spark.using(
    phone_numbers=Input('/examples/phone_numbers'),
    country_codes=Input('/examples/country_codes'),
    output=Output('/examples/phone_numbers_to_country')
)
def map_phone_number_to_country(phone_numbers, country_codes, output):
    # これは前回の実行以降に追加された未読み取りの電話番号すべてになります
    phone_numbers = phone_numbers.dataframe()
    # これは以前に読み取った内容に関係なく、すべての国コードになります
    country_codes = country_codes.dataframe()

    cond = [phone_numbers.country_code == country_codes.code]
    output.write_dataframe(phone_numbers.join(country_codes, on=cond, how='left_outer'))

スナップショット入力の動作は、トランスフォームをインクリメンタルに実行する場合も、非インクリメンタルに実行する場合も同じです。そのため、added および current 読み取りモードは常にデータセット全体を返します。それ以外のすべての読み取りモードは空のデータセットを返します。

入力への変更

既存の入力のリストは変更できます。次のいずれかに該当する場合、インクリメンタル性は維持されます。

• 新しい入力または新しいスナップショット入力が追加された場合。
• 既存の入力または既存のスナップショット入力が削除された場合。

インクリメンタルトランスフォームには、入力が1つ以上必要です。

スナップショット入力以外の各入力データセットの開始トランザクションは、前回の実行で使用されたものと一致している必要があります。

同じトランスフォームによって最後にビルドされた出力

複数出力のインクリメンタルトランスフォームでは、以前にビルドされたすべての出力の最後にコミットされたトランザクションが、同じトランスフォームによって生成されている必要があります。

過去のビルド履歴がない出力は、上記の条件の対象外となり、ビルドのインクリメンタル実行を妨げません。

トランスフォームの入力データセットと出力データセットは異なる必要がある

データのトランスフォームは、入力データセットを受け取り、操作を実行し、出力データセットを生成します。トランスフォームの入力データセットと出力データセットは異なる必要があります。同じデータセットを入力と出力の両方に使用すると、循環依存関係が生じ、トランスフォームを実行できなくなります。

インクリメンタル計算の要件の概要

トランスフォームをインクリメンタルに実行できるのは、すべてのインクリメンタル入力への変更がファイルの追加のみである場合、またはファイルが削除されている場合は、その削除が allow_retention=True を設定した保持処理によって、あるいは allow_user_deletes=True を設定した保持処理以外の方法によって行われている場合に限ります。スナップショット入力はこのチェックの対象外です。