Uso dei motori streaming e GPU

Scalare e ottimizzare le pipeline di dati con Polars

Liam Brannigan

Data Scientist & Polars Contributor

Motori di esecuzione delle query

Diagramma di flusso che confronta una query dall'API, tramite l'ottimizzatore di query, al motore in memoria.

Motori di esecuzione delle query

Diagramma di flusso che confronta una query dall'API, tramite l'ottimizzatore di query, al motore in memoria.

Motori di esecuzione delle query

Diagramma di flusso che confronta una query dall'API, tramite l'ottimizzatore di query, ai motori in memoria e in streaming.

Motori di esecuzione delle query

Diagramma di flusso che confronta una query dall'API, tramite l'ottimizzatore di query, ai motori in memoria, streaming e GPU.

Un riepilogo lazy delle richieste

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

query = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .group_by("DEPARTMENT")
    .len()
    .sort("len", descending=True)
)

Uso del motore predefinito

query.collect()

shape: (5, 2)
| DEPARTMENT        | len     |
| ---               | ---     |
| str               | u32     |
|-------------------|---------|
| 311 City Services | 4859161 |
| Sanitation        | 3406631 |
| Aviation          | 2337842 |
| Transportation    | 1468240 |
| Water             | 416535  |

Uso del motore streaming

query.collect(
    engine="streaming"
)

shape: (5, 2)
| DEPARTMENT        | len     |
| ---               | ---     |
| str               | u32     |
|-------------------|---------|
| 311 City Services | 4859161 |
| Sanitation        | 3406631 |
| Aviation          | 2337842 |
| Transportation    | 1468240 |
| Water             | 416535  |

Uso del motore streaming

Immagine di una grande tabella in streaming.

I passaggi non supportati in streaming tornano automaticamente al motore predefinito

Uso del motore streaming

import polars as pl
pl.Config.set_engine_affinity("streaming")

Uso del motore GPU

query.collect(
    engine="gpu"
)

Richiede hardware NVIDIA GPU

Scelta del motore

Prima il motore predefinito; passa ad altri solo se hai limiti di memoria

Streaming per query grandi, oltre la memoria

GPU per velocità, ma solo con hardware NVIDIA disponibile

Scelta di un motore

Lavorare a batch

completed_requests = (
    requests
    .filter(pl.col("STATUS") == "Completed")
    .select("SR_NUMBER", "TYPE", "DEPARTMENT", "CREATED_DATE")
)

Elaborare ogni batch

for batch in completed_requests.collect_batches(

):

Elaborare ogni batch

for batch in completed_requests.collect_batches(
    chunk_size=50_000,
):

Elaborare ogni batch

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)

Passiamo alla pratica !

Scalare e ottimizzare le pipeline di dati con Polars