注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

Polars 遅延 API

Polars 遅延 API ↗ は、クエリの結果が収集された後にのみクエリを評価します。これにより、エンジンはほとんどの場合にパフォーマンスを向上させる最適化を適用できます。Polars を使用する際には、この計算モードを推奨します。大量のデータを処理するパイプラインには、遅延実行を推奨します。ただし、効果の程度はクエリによって異なり、本番システムにデプロイする前に、個々のパイプラインの安定性を必ず確認する必要があります。

Foundry で Polars 遅延 API にアクセスするには、以下のようにトランスフォームの lazy フラグを True に設定します。

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


@transform.lightweight(
    output=Output("/Users/jsmith/output"),
    input=Input("/Users/jsmith/input"),
)
def compute(output, input):
    df = input.polars(lazy=True)
    # データ変換ロジック
    output.write_table(df)

デフォルトでは、遅延実行は無効です。

遅延計算では、通常、Polars のストリーミングエンジン ↗が使用されます。クエリを一度にまとめて実行する代わりにバッチ単位で実行できるため、メモリー容量を超えるデータも処理できます。また、データストリーミングを有効にすると、クエリの実行が高速になります。Polars を遅延モードで使用する場合、ストリーミングは常に有効です。

フィルタープッシュダウン

遅延実行は、述語プッシュダウンとも呼ばれるフィルタープッシュダウンを使用できる場合に特に有効です。フィルタープッシュダウンでは、.filter()、.where()、ブールインデックス化などの絞り込み処理は即座に実行されず、クエリプランの一部として記録されます。クエリの実行時に、Polars はこれらのフィルターを実行プラン内のできるだけ早い段階に移動しようとします。場合によっては、データスキャンそのものにまでフィルターを移動でき、データの入出力を完全に回避できます。パイプラインで使用するデータの割合が小さいほど、フィルタープッシュダウンの効果は大きくなります。遅延計算で使用される最適化のすべてについては、Polars の遅延最適化 ↗のドキュメントを参照してください。

クエリが以下の例に類似している場合は、パフォーマンスが大幅に向上するため、遅延実行を有効にすることを強く推奨します。

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


@transform.lightweight(
    output=Output("/Users/jsmith/2025_06_sale_data"),
    sales_data=Input("/Users/jsmith/sales_data"),
)
def june_2025_sales(output, sales_data):
    df = (
        sales_data.polars(lazy=True)
        .filter(
            (pl.col("date") >= pl.lit("2025-06-01")) &
            (pl.col("date") <= pl.lit("2025-06-30"))
        )
    )
    output.write_table(df)

遅延パイプラインをデバッグする

遅延実行モードでは、Polars がクエリプランを生成します。遅延 DataFrame に対して .explain() または .describe_plan() を使用して、予定されている実行ステップと適用された最適化を表示します。

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


@transform.lightweight(
    output=Output("/Users/jsmith/output"),
    input=Input("/Users/jsmith/input"),
)
def compute(output, input):
    df = input.polars(lazy=True)
    # データ変換ロジック
    print(df.explain())
    output.write_table(df)

クエリの実行に失敗し、遅延実行によって問題の特定が難しくなる場合は、.collect() で中間結果を実体化できます。

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


@transform.lightweight(
    output=Output("/Users/jsmith/output"),
    input=Input("/Users/jsmith/input"),
)
def compute(output, input):
    df = input.polars(lazy=True)
    # クエリの一部
    df.collect()
    # クエリの残り
    output.write_table(df)

ストリーミングデータセットに対するインクリメンタルトランスフォームで使用する

ストリーミングデータセットに対するインクリメンタル（差分処理）トランスフォームで polars(lazy=True) を使用する場合、インクリメンタルウィンドウに新しいファイルが含まれていないことがあります。この場合、空の入力に対して .polars(lazy=True) を呼び出すと、BinderException が発生することがあります。これに対処するには、遅延 API を使用する前に入力が空かどうかを確認します。

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)

この既知の制限の詳細については、インクリメンタルトランスフォームのドキュメントを参照してください。

GPU アクセラレーション

GPU を基盤とする実行エンジン cudf-polars を使用すると、GPU 処理によって Polars 遅延 API のクエリを高速化できます。collect() 呼び出しに pl.GPUEngine() を渡して、既存の Polars 遅延 API のクエリを GPU 処理で実行します。

Copied!1
result = df.collect(engine=pl.GPUEngine())

GPU アクセラレーションによる結果の収集は、結果全体をメモリー内に実体化する collect() に対応しています。sink_parquet() などのストリーミング処理には対応していません。

詳細なセットアップ手順、対応している処理、およびコード例については、cuDF による GPU アクセラレーション対応 Polars を参照してください。

遅延 API を使用するタイミング

遅延計算における Polars の最適化を最大限に活用できるように、Foundry データセットは内部オブジェクトストアプロキシを介して公開されます。そのため、クエリは必要なデータのみを読み込みます。即時モードの Polars には同じ最適化がないため、lazy=False の場合はデータセット全体がディスクにプリフェッチされます。

遅延実行を有効にするタイミングを判断する際は、以下の点を考慮します。

以下の場合は遅延実行（lazy=True）を使用します。
- 大規模なデータセットまたは複雑な変換パイプラインを扱う場合。
- フィルタープッシュダウン、述語プッシュダウン、射影プッシュダウン、ストリーミングなどのクエリ最適化を活用したい場合。
- 複数の処理が連鎖しており、Polars に実行順序を最適化させたい場合。
- メモリーに収まらないデータを処理する必要がある場合。
以下の場合は即時実行（lazy=False）を使用します。
- プロトタイプを作成している場合、またはデータ変換を段階的に探索している場合。
- 複雑なパイプラインをデバッグしており、より予測しやすい動作を求める場合。