Lucrul cu motoarele streaming și GPU

Scalarea și optimizarea pipeline-urilor de date cu Polars

Liam Brannigan

Data Scientist & Polars Contributor

Motoare de execuție a interogărilor

Diagramă de flux comparând o interogare de la API, prin optimizatorul de interogări, la motorul in-memory.

Scalarea și optimizarea pipeline-urilor de date cu Polars

Motoare de execuție a interogărilor

Diagramă de flux comparând o interogare de la API, prin optimizatorul de interogări, la motorul in-memory.

Scalarea și optimizarea pipeline-urilor de date cu Polars

Motoare de execuție a interogărilor

Diagramă de flux comparând o interogare de la API, prin optimizatorul de interogări, la motoarele in-memory și streaming.

Scalarea și optimizarea pipeline-urilor de date cu Polars

Motoare de execuție a interogărilor

Diagramă de flux comparând o interogare de la API, prin optimizatorul de interogări, la motoarele in-memory, streaming și GPU.

Scalarea și optimizarea pipeline-urilor de date cu Polars

Un rezumat lazy al cererilor

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

query = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .group_by("DEPARTMENT")
    .len()
    .sort("len", descending=True)
)
Scalarea și optimizarea pipeline-urilor de date cu Polars

Utilizarea motorului implicit

query.collect()

shape: (5, 2)
| DEPARTMENT        | len     |
| ---               | ---     |
| str               | u32     |
|-------------------|---------|
| 311 City Services | 4859161 |
| Sanitation        | 3406631 |
| Aviation          | 2337842 |
| Transportation    | 1468240 |
| Water             | 416535  |
Scalarea și optimizarea pipeline-urilor de date cu Polars

Selectarea motorului streaming

query.collect(
    engine="streaming"
)

shape: (5, 2)
| DEPARTMENT        | len     |
| ---               | ---     |
| str               | u32     |
|-------------------|---------|
| 311 City Services | 4859161 |
| Sanitation        | 3406631 |
| Aviation          | 2337842 |
| Transportation    | 1468240 |
| Water             | 416535  |
Scalarea și optimizarea pipeline-urilor de date cu Polars

Selectarea motorului streaming

Imagine a unui tabel mare procesat în streaming.

  • Pașii nesupportați în streaming revin automat la motorul implicit
Scalarea și optimizarea pipeline-urilor de date cu Polars

Selectarea motorului streaming

import polars as pl
pl.Config.set_engine_affinity("streaming")
Scalarea și optimizarea pipeline-urilor de date cu Polars

Selectarea motorului GPU

query.collect(
    engine="gpu"
)

$$

  • Necesită hardware NVIDIA GPU
Scalarea și optimizarea pipeline-urilor de date cu Polars

Alegerea unui motor

$$

  • Mai întâi motorul implicit; comutați doar dacă memoria este o problemă

$$

  • Streaming pentru interogări mari, care depășesc memoria

$$

  • GPU pentru viteză, doar dacă este disponibil hardware NVIDIA

Alegerea unui motor

Scalarea și optimizarea pipeline-urilor de date cu Polars

Procesarea în loturi

completed_requests = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .select("SR_NUMBER", "TYPE", "DEPARTMENT", "CREATED_DATE")
)
Scalarea și optimizarea pipeline-urilor de date cu Polars

Procesarea fiecărui lot

for batch in completed_requests.collect_batches(

):
Scalarea și optimizarea pipeline-urilor de date cu Polars

Procesarea fiecărui lot

for batch in completed_requests.collect_batches(
    chunk_size=50_000,
):
Scalarea și optimizarea pipeline-urilor de date cu Polars

Procesarea fiecărui lot

for batch in completed_requests.collect_batches(
    chunk_size=50_000,
):
    print(batch.shape)
    batch_json = batch.write_json()

(50000, 4)
(50000, 4)
...
(50000, 4)
(12409, 4)
Scalarea și optimizarea pipeline-urilor de date cu Polars

Să exersăm!

Scalarea și optimizarea pipeline-urilor de date cu Polars

Preparing Video For Download...