インクリメンタル（差分処理）トランスフォーム

インクリメンタル計算は、入力データセットにトランスフォームを実行して出力データセットを生成する効率的な方法です。トランスフォームのビルド履歴を活用することで、トランスフォームを実行するたびに出力データセット全体を再計算する必要がなくなります。入力データセットと出力データセットを同じにすることはできません。循環依存関係の作成を避けるため、入力データセットと出力データセットは異なる必要があります。

インクリメンタルパイプラインの作成と管理に関する一連のガイダンスについては、パイプラインの構築セクションを参照してください。

例：非インクリメンタルトランスフォームとインクリメンタルトランスフォーム

このセクションでは、まずインクリメンタルトランスフォームを使用しないコードの例を取り上げ、インクリメンタルトランスフォームの利点を説明します。

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


@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
10
11
12
from transforms.api import transform, Input, Output


@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[students_df['hair'] == 'Brown'])

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

@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
    query = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")
    processed.write_table(query)

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


@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 に新しいデータが追加されると、入力に追加された新しいデータだけでなく、入力全体に対して filter() が実行されます。これは計算リソースと時間の両方を無駄にします。

トランスフォームが自身のビルド履歴を把握できれば、出力をより効率的に計算できます。具体的には、入力に加えられた変更を使用して、出力データセットを変更できます。このように、テーブルを再実体化するときに、すでに実体化されたデータを使用する処理をインクリメンタル計算と呼びます。インクリメンタル計算を使用しない場合、出力データセットは常に最新のトランスフォームの出力で置き換えられます。

上記のトランスフォームの例に戻ります。このトランスフォームは、students データセットに対して filter() を実行し、茶色の髪の学生を出力します。インクリメンタル計算を使用すると、2人の新しい学生のデータが students に追加された場合、トランスフォームはビルド履歴の情報を使用して、新しく追加された茶色の髪の学生だけを出力に追加できます。

+---+-----+-----+------+                  +---+-----+-----+------+
| id| hair|  eye|   sex|                  | id| hair|  eye|   sex|
+---+-----+-----+------+     Build 1      +---+-----+-----+------+
| 17|Black|Green|Female|    --------->    | 18|Brown|Green|Female|
| 18|Brown|Green|Female|                  +---+-----+-----+------+
| 19|  Red|Black|Female|
+---+-----+-----+------+
        ...                                         ...
+---+-----+-----+------+     Build 2      +---+-----+-----+------+
| 20|Brown|Amber|Female|    --------->    | 20|Brown|Amber|Female|
| 21|Black|Blue |Male  |                  +---+-----+-----+------+
+---+-----+-----+------+

したがって、上記のトランスフォームの例は、次の構文でインクリメンタルロジックを使用するように書き換えられます。

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('added')
    processed.write_table(students_df.filter(pl.col('hair') == 'Brown'))

Copied!1
2
3
4
5
6
7
8
9
10
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.pandas('added')
    processed.write_table(students_df[students_df['hair'] == 'Brown'])

Copied!1
2
3
4
5
6
7
8
9
10
11
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(read_modes={students: "added"}).conn
    query = conn.sql("SELECT * FROM students WHERE hair = 'Brown'")
    processed.write_table(query)

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 デコレーターの詳細については、インクリメンタルトランスフォームのリファレンスを参照してください。

制限事項と考慮事項

Code Repositories のプレビュー機能は、incremental() デコレーターに require_incremental=True を渡した場合でも、常に非インクリメンタルモードでトランスフォームを実行します。VS Code のプレビューでは、インクリメンタルの動作が反映されます。

一方のデータセットが全体を読み込む参照テーブルで、もう一方が差分だけを読み込むインクリメンタルデータセットである場合、結合を安全に計算できます。ただし、結合する両方のデータセットをインクリメンタルに読み込む場合は、特別な処理が必要です。詳細については、インクリメンタルトランスフォームを活用して大規模なデータセットを結合する例を参照してください。

ストリーミングデータセットでインクリメンタルな軽量トランスフォームを実行すると、インクリメンタルウィンドウに処理対象の新しいファイルが含まれていない場合があります。基盤となる統合が空でないファイルリストを想定しているため、空の入力で .polars(lazy=True) を呼び出すと BinderException が発生する可能性があります。このエラーを避けるには、コンテキストの is_incremental プロパティを使用し、.polars(lazy=True) を呼び出す前に入力にデータがあるかどうかを確認します。

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


@incremental()
@transform.using(
    my_output=Output('/path/to/output'),
    my_input=Input('/path/to/input'),
)
def compute(ctx, my_output, my_input):
    if ctx.is_incremental and my_input.polars().is_empty():
        return
    df = my_input.polars(lazy=True)
    my_output.write_table(df)

または、トランスフォームを実行する前に、インクリメンタルウィンドウに常に1つ以上のファイルが含まれるようにすることもできます。遅延 API の詳細については、Polars 遅延 API のドキュメントを参照してください。

メディアセットで使用する

メディアセットでもインクリメンタル計算がサポートされるようになりました。詳細については、インクリメンタルメディアセットのドキュメントを参照してください。

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

メディアセットの入力を含むインクリメンタルトランスフォームで、トランザクション型データセットのバッチ処理を設定できます。詳細については、インクリメンタル入力のバッチサイズの制限に関するドキュメントを参照してください。メディアセットについては、インクリメンタルメディアセット入力のバッチサイズの制限を参照してください。

←

PREVIOUS基本的なトランスフォーム / DuckDB API

NEXT使用ガイド

→