Transférer de grosses requêtes sur disque

Mise à l'échelle et optimisation des pipelines de données avec Polars

Liam Brannigan

Data Scientist & Polars Contributor

Un problème de grand résultat

Schéma montrant une requête paresseuse produisant un grand extrait préservant les lignes, à écrire directement sur disque.

Mise à l'échelle et optimisation des pipelines de données avec Polars

Un jeu de requêtes nettoyé

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

clean_requests = (
    requests





)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Un jeu de requêtes nettoyé

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"),
    )
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Collecter puis écrire

clean_requests.collect(
    engine="streaming"
).write_parquet(
    "311_requests_clean.parquet"
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Écrire directement en Parquet

clean_requests.sink_parquet(
    "311_requests_clean.parquet",


)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Paramétrer l'écriture

clean_requests.sink_parquet(
    "311_requests_clean.parquet",
    compression="zstd",
    row_group_size=100_000,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Autres méthodes d'écriture

df.sink_csv("output.csv")
df.sink_ndjson("output.ndjson")
Mise à l'échelle et optimisation des pipelines de données avec Polars

Sortie partitionnée

clean_requests.sink_parquet(
    pl.PartitionBy(



    ),

)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Sortie partitionnée

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

)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Sortie partitionnée

clean_requests.sink_parquet(
    pl.PartitionBy(
        "311_requests_clean/",
        key="CREATED_DATE",
        max_rows_per_file=1_000_000,
    ),
    mkdir=True,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Sortie partitionnée

311_requests_clean/
  CREATED_DATE=2025-12-31/
    00000000.parquet
  CREATED_DATE=2026-01-01/
    00000000.parquet
Mise à l'échelle et optimisation des pipelines de données avec Polars

Partitions sur plusieurs colonnes

clean_requests.sink_parquet(
    pl.PartitionBy(
        "311_requests_clean/",
        key=["STATUS", "CREATED_DATE"],
    ),
    mkdir=True,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Construire des écritures paresseuses

internal_sink = clean_requests.sink_parquet(
    "311_requests_internal.parquet",
    lazy=True,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Construire des écritures paresseuses

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,
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Multiplexer les écritures

pl.collect_all(
    [internal_sink, public_sink],
)
Mise à l'échelle et optimisation des pipelines de données avec Polars

Fichiers produits

311_requests_internal.parquet
311_requests_public.parquet
Mise à l'échelle et optimisation des pipelines de données avec Polars

Fonctions d'écriture personnalisées par lot

def send_batch(batch: pl.DataFrame) -> None:
    print(batch.height)
    batch_json = batch.write_json()
    # send batch_json to an API
Mise à l'échelle et optimisation des pipelines de données avec Polars

Fonctions d'écriture personnalisées par lot

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

50000
50000
...
29268
Mise à l'échelle et optimisation des pipelines de données avec Polars

Flux de travail d'écriture

  • Utilisez sink_parquet() pour écrire de gros résultats de requête directement sur disque
  • Utilisez PartitionBy() pour scinder la sortie en ensemble de données partitionné
  • Utilisez des écritures paresseuses avec collect_all() pour planifier et exécuter plusieurs sorties ensemble
  • Utilisez sink_batches() quand chaque lot doit passer par une fonction personnalisée

Vue isométrique d'un système de tri distribuant les données dans des tiroirs partitionnés et organisés

Mise à l'échelle et optimisation des pipelines de données avec Polars

Passons à la pratique !

Mise à l'échelle et optimisation des pipelines de données avec Polars

Preparing Video For Download...