Streaming con ejecución diferida en Polars: cuándo usar collect o sink_parquet

Imaginemos a un desarrollador que trabaja con una única tabla de origen. Un analista necesita un pequeño resumen por categoría en Python; una canalización de conjuntos de datos necesita guardar determinados registros en disco. La misma entrada, tareas diferentes. El destino debe orientar la elección de cómo ejecutar la consulta.

Para el procesamiento en streaming con ejecución diferida de Polars, empieza con una lectura mediante scan desde un archivo e inspecciona la consulta. Usa collect cuando el consumidor necesite un DataFrame cuyo tamaño se haya limitado deliberadamente, o sink_parquet cuando el resultado deba guardarse en disco. La guía de orígenes y destinos explica cómo se conectan la lectura y el procesamiento con la escritura. El streaming no garantiza un límite máximo de memoria.

Esta guía sigue la documentación. El ejemplo con un archivo local es hipotético y no se ha ejecutado, y no hay resultados de pruebas de rendimiento.

Empieza con una lectura mediante scan, antes de cargar un DataFrame

Qué permiten ver scan_parquet y scan_csv al optimizador

Un LazyFrame basado en un archivo describe el trabajo antes de materializar el resultado. Con scan_parquet o scan_csv, el optimizador puede acercar al lector los filtros y las selecciones de columnas que lo permitan. La guía de la API de ejecución diferida ilustra cómo se trasladan los predicados y las proyecciones hacia el origen mediante una agregación con filtros.

La proyección conserva las columnas que necesita una consulta. El traslado de predicados hacia el origen acerca las condiciones de filas adecuadas a la fuente de datos. Ninguna de estas técnicas permite cambiar lo que recibe el consumidor. Si un proceso posterior necesita un identificador, eliminarlo cambia el contrato de datos. También lo cambia descartar valores ausentes que tienen significado.

No todas las expresiones pueden trasladarse al lector. CSV y Parquet no tienen por qué ofrecer el mismo ahorro de E/S. Inspecciona el plan y después mide la E/S antes de afirmar que hay ahorros.

Por qué read().lazy() no puede deshacer una lectura inmediata

Cuando read_parquet(...).lazy() llega a .lazy(), la carga inmediata ya se ha producido. Las transformaciones posteriores pueden ser diferidas; la lectura inicial no puede deshacerse. La documentación de orígenes y destinos recomienda scan_* en lugar de read_* para flujos de trabajo con archivos que deban usar ejecución diferida.

Eso no significa que usar un DataFrame con ejecución inmediata sea un error. La guía de la API de ejecución diferida describe su utilidad durante la exploración, cuando los resultados intermedios ayudan a comprender los datos. Elígelo por ese motivo, en vez de reescribir un trabajo exploratorio útil solo para llamarlo streaming.

El ejemplo presupone un archivo local hipotético llamado events.parquet. Sus columnas record_id y category son de tipo String; value es de tipo Float64. No se ha inspeccionado ningún esquema. Los dos posibles consumidores necesitan registros seleccionados a nivel de fila o un resumen por categoría derivado de ellos.

Lee por separado el plan lógico y el plan físico de streaming

Usa explain para inspeccionar la optimización

Compara query.explain(optimized=False) con query.explain(optimized=True). La guía de planes de consulta muestra cómo un filtro se traslada a una lectura mediante scan tras la optimización. Estas vistas describen el trabajo previsto. No miden el tiempo de ejecución ni el pico de memoria, y tampoco establecen la corrección del resultado.

Comprueba que aparezcan la proyección y la selección esperadas. Si falta alguna, investiga antes de interpretar las mediciones de recursos. Después, comprueba que el propio filtro cumpla la regla de negocio: un filtro situado en el lugar correcto puede seguir seleccionando las filas equivocadas.

Usa el grafo físico para investigar el trabajo sensible al consumo de memoria

La guía de streaming documenta query.show_graph(plan_stage="physical", engine="streaming") y una leyenda para las operaciones que pueden consumir mucha memoria. Para representar el grafo, Graphviz debe estar disponible en PATH, como explica la guía de planes de consulta. Sin él, la salida textual de explain sigue siendo útil, pero no proporciona la misma vista física.

El streaming trabaja por lotes. Algunas operaciones son inherentemente incompatibles con el streaming; otras aún no tienen una implementación de streaming y recurren a la ejecución en memoria. Usa el grafo físico para detectar esos aspectos. No garantiza un límite del pico de RSS.

Comprueba también los valores predeterminados de cada versión. La referencia estable actual de collect indica que engine="auto" sigue Config.set_engine_affinity o POLARS_ENGINE_AFFINITY y, en su ausencia, usa streaming de forma predeterminada. Las instalaciones anteriores pueden comportarse de otra manera. Especificar engine="streaming" expresa tu intención, pero no demuestra que todas las operaciones se ejecuten en streaming. Registra la versión instalada y la configuración antes de basarte en la documentación actual.

Elige collect o sink_parquet según la salida que necesites

Pregúntate qué requiere el siguiente consumidor. El tamaño del archivo de origen por sí solo no puede responder a esa pregunta.

Un resultado de tamaño limitado en memoria

collect materializa un LazyFrame en un DataFrame. El objeto devuelto necesita memoria aunque el trabajo anterior se ejecute en streaming. La agregación por sí sola no lo hace pequeño: el número de categorías determina cuántas filas quedan en el resumen.

Un resultado persistido sin materializar toda la salida con collect

La referencia de sink_parquet describe la evaluación en streaming y la escritura en Parquet sin exigir primero que quien realiza la llamada obtenga el DataFrame final completo. El valor predeterminado lazy=False ejecuta la escritura en el destino. Con lazy=True, la ejecución espera a que se invoque collect; los argumentos engine y optimizations de la operación de escritura no se aplican al construir ese plan diferido. Elige los ajustes de ejecución al invocar collect sobre él. La opción diferida está marcada como inestable.

Distingue las reglas de ejecución alternativa de ambos métodos. La referencia actual de sink sigue la afinidad de motor para la selección automática y, en su ausencia, usa streaming; también describe el recurso al streaming como alternativa al motor seleccionado. La referencia de collect describe el recurso a la ejecución en memoria como alternativa al motor seleccionado. Los límites de cada operación siguen siendo aplicables. Los ajustes de compresión y de grupos de filas no establecen un presupuesto fijo de memoria.

La pregunta histórica de la comunidad sobre DuckDB y Polars plantea la captura de cambios en los datos de PostgreSQL. Léela como una pregunta terminológica, no como asesoramiento de arquitectura verificado. El streaming por lotes descrito aquí no proporciona un servicio siempre activo con posiciones de lectura, tratamiento del tiempo de los eventos o garantías de entrega.

La guía práctica de Polars centrada en el resultado

Define el contrato del resultado antes de elegir la ejecución

La guía práctica de Polars centrada en el resultado es una lista de comprobación editorial original, no una biblioteca ni un despliegue validado. Empieza por identificar al consumidor. Acuerda las columnas y los tipos necesarios y, después, establece los filtros, el tratamiento de valores nulos y cualquier requisito de ordenación. Esas decisiones te dan algo concreto que validar tras la ejecución.

El diagrama muestra conceptualmente la elección del destino. No es un plan físico de Polars observado.

Lee mediante scan, reduce e inspecciona

Este ejemplo presupone que Polars ya está disponible y utiliza la API estable documentada actualmente. No se ha ejecutado. No se han comprobado la configuración ni el contenido del conjunto de datos, y no se afirma que exista ninguna salida impresa. El filtro descarta las filas con value nulo únicamente para ilustrar una política; tu conjunto de datos puede requerir otra.

import polars as pl

rows = (
    pl.scan_parquet("events.parquet")
    .select("record_id", "category", "value")
    .filter(pl.col("value").is_not_null())
)

planned_schema = rows.collect_schema()
unoptimized_plan = rows.explain(optimized=False)
optimized_plan = rows.explain(optimized=True)

collect_schema resuelve el esquema previsto. Esto puede leer metadatos del disco o realizar solicitudes remotas, por lo que la resolución del esquema no está necesariamente exenta de coste. Tampoco valida cada fila según las reglas del dominio. Compara los nombres y los tipos con el contrato y después inspecciona el plan lógico y, cuando las herramientas lo permitan, el grafo físico. La guía de esquemas explica por qué es importante conocer los esquemas durante la planificación de la ejecución diferida.

Elige uno de los siguientes métodos para la salida necesaria. Son alternativas que no se han ejecutado, no una secuencia. La rama del resumen presupone que has establecido de forma independiente que el número de categorías es manejable. También debes decidir qué significa una categoría nula.

La alternativa A produce el resumen por categoría necesario en Python.

summary = rows.group_by("category").agg(
    pl.col("value").mean().alias("mean_value")
)
result = summary.collect(engine="streaming")

La alternativa B escribe en disco los registros seleccionados necesarios.

rows.sink_parquet("events-selected-candidate.parquet", engine="streaming")

La API de collect y la API de sink respaldan estas descripciones de los métodos finales. Comparar sus tiempos supondría comparar tareas diferentes: uno devuelve un resumen por categoría, mientras que el otro exporta registros a nivel de fila. Eso no es una prueba de rendimiento del motor.

Propón una comprobación comparable y después valida el destino

Una comparación útil mantiene constante la salida. Usa la misma instantánea de entrada y la misma versión de Polars, con ajustes, equipo y condiciones de almacenamiento comparables. La guía de Optijara sobre mantener los protocolos de las pruebas de rendimiento vinculados a los resultados aborda esa disciplina en otro ámbito. Mantén separado el protocolo propuesto de los resultados que registres finalmente.

Aquí no se ha realizado ninguna de estas mediciones. Para un candidato persistido, comprueba el esquema y los recuentos de filas sin cargar toda la salida únicamente para validarla. Compara los valores con las tolerancias acordadas; compara el orden solo cuando el contrato lo exija.

Si la ejecución falla, considera los archivos parciales como no validados. Investiga el error y las operaciones implicadas. Usa un nuevo destino candidato para el siguiente intento y valídalo antes de cualquier entrega aprobada. La presencia de un archivo no es una comprobación de aceptación.

Errores de los equipos al usar streaming con ejecución diferida

El atajo tentador consiste en tratar la planificación diferida y la ejecución en streaming como una única decisión. Son decisiones distintas. Añadir .lazy() después de una lectura no puede cambiar esa carga ya completada. Elegir streaming no elimina el DataFrame final que devuelve collect.

La inspección también puede parecer más concluyente de lo que es. Un plan explica el trabajo previsto; un esquema resuelto proporciona nombres y tipos. Ninguno responde a si el uso de memoria es aceptable o si las filas conservan el significado previsto. Mantén esas comprobaciones separadas de la inspección.

Una comparación de rendimiento resulta difícil de interpretar si cambias los filtros o la proyección entre pruebas. Cambiar la instantánea o el destino genera el mismo problema. Registra el cambio de tarea en lugar de atribuir todas las diferencias al motor. Y trata los requisitos de ingesta continua aparte de la decisión sobre un método por lotes.

Para las filas extraídas de PDF, comprueba que los encabezados y las unidades sigan correspondiendo a los valores correctos, y conserva las referencias de origen antes de la transformación tabular. La guía de Optijara sobre conservar el significado de las tablas extraídas de PDF y las referencias de origen cubre ese trabajo previo. No establece una integración con Polars. Un destino de escritura diferente no puede recuperar el significado perdido durante la extracción.

Límites que debes comprobar antes de cambiar una canalización de datos

Consideraciones sobre memoria, versión y almacenamiento

El tamaño de entrada es solo una parte de la cuestión de la memoria. Ten en cuenta el estado intermedio y el resultado final, y comprueba qué operaciones pueden ejecutarse en streaming. La documentación de streaming y de los métodos finales explica esos límites. La adopción también requiere esfuerzo de implementación y añade E/S de almacenamiento o complejidad operativa. Aquí no se promete ninguna mejora del tiempo de ejecución ni de los costes.

La investigación de PythonSpeed sobre memoria es un ejemplo histórico concreto de análisis de una carga de trabajo específica. Sus mediciones se aplican a su conjunto de datos, sus versiones y su plataforma. No son resultados actuales universales ni pruebas de Optijara.

Antes de usar valores predeterminados o parámetros opcionales, comprueba la documentación de tu versión instalada. La referencia de sink marca varias opciones como inestables o experimentales. Si quieres comprender un cambio, evita ajustar la compresión y la ordenación al mismo tiempo que los ajustes de ejecución.

El significado de los datos y la privacidad siguen necesitando revisión

Durante la preparación, conserva los identificadores y los límites entre las particiones del conjunto de datos. Haz explícito el tratamiento de los valores nulos y conserva la ordenación cuando el consumidor la requiera. Un archivo Parquet válido no demuestra que se hayan conservado los ejemplos previstos. La guía relacionada sobre conservar la selección del conjunto de datos y la identidad temporal explica esas comprobaciones para conjuntos de datos que dependen del tiempo; no afirma que exista una integración con Polars.

Usa instantáneas de entrada autorizadas. Decide quién puede acceder a las salidas candidatas y cuánto tiempo conservarlas. Un destino local no resuelve las obligaciones de privacidad, aunque la consulta se ejecute correctamente.

Elige el destino antes de invocar collect

Empieza por el resultado que necesita el consumidor. Usa una lectura mediante scan para el trabajo basado en archivos que deba emplear ejecución diferida y después inspecciona los planes lógico y físico según sus distintas funciones. Valida el método final elegido y conserva las pruebas de medición reales antes de cambiar una canalización operativa. Para ayudarte a planificar un flujo de trabajo de IA, Optijara ofrece asesoramiento en IA.