大規模クエリのディスクへの書き出し

Polars によるデータパイプラインのスケーリングと最適化

Liam Brannigan

Data Scientist & Polars Contributor

大量出力の問題

大量の行を保持するクエリ結果をディスクへ直接書き出す流れを示す図

クリーニング済みデータセット

requests = pl.scan_parquet("311_Service_Requests.parquet")

clean_requests = (
    requests





)

クリーニング済みデータセット

requests = pl.scan_parquet("311_Service_Requests.parquet")

clean_requests = (
    requests
    .with_columns(
        pl.col("CREATED_DATE").dt.date(),
        pl.col("WARD").cast(pl.Int16),
        (pl.col("STATUS") == "Completed").alias("IS_COMPLETED"),
    )
)

収集してから書き出す

clean_requests.collect(
    engine="streaming"
).write_parquet(
    "311_requests_clean.parquet"
)

Parquet へのシンク

clean_requests.sink_parquet(
    "311_requests_clean.parquet",


)

シンクの制御

clean_requests.sink_parquet(
    "311_requests_clean.parquet",
    compression="zstd",
    row_group_size=100_000,
)

その他のシンクメソッド

df.sink_csv("output.csv")
df.sink_ndjson("output.ndjson")

パーティション出力

clean_requests.sink_parquet(
    pl.PartitionBy(



    ),

)

パーティション出力

clean_requests.sink_parquet(
    pl.PartitionBy(
        "311_requests_clean/",
        key="CREATED_DATE",
        max_rows_per_file=1_000_000,
    ),

)

パーティション出力

clean_requests.sink_parquet(
    pl.PartitionBy(
        "311_requests_clean/",
        key="CREATED_DATE",
        max_rows_per_file=1_000_000,
    ),
    mkdir=True,
)

パーティション出力

311_requests_clean/
  CREATED_DATE=2025-12-31/
    00000000.parquet
  CREATED_DATE=2026-01-01/
    00000000.parquet

複数列パーティション

clean_requests.sink_parquet(
    pl.PartitionBy(
        "311_requests_clean/",
        key=["STATUS", "CREATED_DATE"],
    ),
    mkdir=True,
)

遅延シンクの構築

internal_sink = clean_requests.sink_parquet(
    "311_requests_internal.parquet",
    lazy=True,
)

遅延シンクの構築

internal_sink = clean_requests.sink_parquet(
    "311_requests_internal.parquet",
    lazy=True,
)

public_sink = clean_requests.drop("REQUEST_ID","REPORTER").sink_parquet(
    "311_requests_public.parquet",
    lazy=True,
)

シンクの多重化

pl.collect_all(
    [internal_sink, public_sink],
)

書き出し結果

311_requests_internal.parquet
311_requests_public.parquet

カスタムバッチシンク

def send_batch(batch: pl.DataFrame) -> None:
    print(batch.height)
    batch_json = batch.write_json()
    # send batch_json to an API

カスタムバッチシンク

clean_requests.sink_batches(
    send_batch,
    chunk_size=50_000,
)

シンクのワークフロー

sink_parquet() で大規模なクエリ結果をディスクへ直接書き出す

PartitionBy() で出力をパーティション分割されたデータセットに分割する

遅延シンクと collect_all() で複数の出力をまとめて計画・実行する

各バッチをカスタム関数に渡す場合は sink_batches() を使用する

データを整理されたパーティション引き出しに振り分けるデータソートシステムの等角投影図

練習しましょう！

Polars によるデータパイプラインのスケーリングと最適化