Trabajar con los motores de streaming y GPU

Escala y optimiza canalizaciones de datos con Polars

Liam Brannigan

Data Scientist & Polars Contributor

Motores de ejecución de consultas

Diagrama de flujo que compara una consulta pasando por la API, el optimizador de consultas y el motor en memoria.

Motores de ejecución de consultas

Diagrama de flujo que compara una consulta pasando por la API, el optimizador de consultas y el motor en memoria.

Motores de ejecución de consultas

Diagrama de flujo que compara una consulta pasando por la API, el optimizador de consultas y los motores en memoria y streaming.

Motores de ejecución de consultas

Diagrama de flujo que compara una consulta pasando por la API, el optimizador de consultas y los motores en memoria, streaming y GPU.

Resumen perezoso de solicitudes

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

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

Usar el motor predeterminado

query.collect()

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

Usar el motor de streaming

query.collect(
    engine="streaming"
)

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

Usar el motor de streaming

Imagen de una tabla grande transmitida.

Pasos no compatibles con streaming vuelven automáticamente al motor predeterminado

Usar el motor de streaming

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

Usar el motor de GPU

query.collect(
    engine="gpu"
)

Requiere hardware NVIDIA GPU

Elegir un motor

Primero el motor predeterminado; cambia solo si hay limitación de memoria

Streaming para consultas grandes que no caben en memoria

GPU para más velocidad, solo si hay hardware NVIDIA

Elegir un motor

Trabajar por lotes

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

Procesar cada lote

for batch in completed_requests.collect_batches(

):

Procesar cada lote

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

Procesar cada lote

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)

¡Vamos a practicar!

Escala y optimiza canalizaciones de datos con Polars