Inviare query grandi su disco

Scalare e ottimizzare le pipeline di dati con Polars

Liam Brannigan

Data Scientist & Polars Contributor

Un problema di output grande

Diagramma che mostra una query lazy che produce un estratto grande, con lo stesso numero di righe, da scrivere direttamente su disco.

Un dataset di richieste pulito

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

clean_requests = (
    requests





)

Un dataset di richieste pulito

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"),
    )
)

Prima collect, poi write

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

Sink in Parquet

clean_requests.sink_parquet(
    "311_requests_clean.parquet",


)

Controllare il sink

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

Altri metodi di sink

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

Output partizionato

clean_requests.sink_parquet(
    pl.PartitionBy(



    ),

)

Output partizionato

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

)

Output partizionato

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

Output partizionato

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

Partizioni multi-colonna

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

Creare sink lazy

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

Creare sink lazy

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,
)

Multiplexing dei sink

pl.collect_all(
    [internal_sink, public_sink],
)

Output scritti

311_requests_internal.parquet
311_requests_public.parquet

Sink a batch personalizzati

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

Sink a batch personalizzati

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

Workflow dei sink

Usa sink_parquet() per scrivere grandi risultati direttamente su disco

Usa PartitionBy() per dividere l'output in un dataset partizionato

Usa sink lazy con collect_all() per pianificare ed eseguire più output insieme

Usa sink_batches() quando ogni batch va a una funzione personalizzata

Vista isometrica di un sistema di ordinamento dati che distribuisce i dati in cassetti partizionati e ordinati

Ayo berlatih!

Scalare e ottimizzare le pipeline di dati con Polars