Polars: execução diferida e em fluxo, quando usar collect ou sink_parquet

Considere um programador hipotético que trabalha a partir de uma única tabela de origem. Um analista quer um pequeno resumo por categoria em Python; um fluxo de preparação de conjuntos de dados precisa de guardar registos selecionados em disco. A mesma entrada, tarefas diferentes. O destino deve orientar a escolha da execução.

Para a execução diferida e em fluxo no Polars, comece com uma leitura diferida de um ficheiro e inspecione a consulta. Use collect quando o consumidor precisar de um DataFrame com limites deliberadamente definidos, ou sink_parquet quando o resultado se destinar ao disco. O guia de origens e destinos explica como a leitura e o processamento se ligam à escrita. O processamento em fluxo não garante um limite máximo de memória.

Este guia segue a documentação. O exemplo com ficheiros locais é hipotético e não foi executado, e não há resultados de testes de desempenho.

Comece com uma leitura diferida, não com um DataFrame já carregado

O que scan_parquet e scan_csv permitem ao otimizador analisar

Um LazyFrame baseado num ficheiro descreve o trabalho antes de o resultado ser materializado. Com scan_parquet ou scan_csv, o otimizador pode aproximar do leitor os filtros e as seleções de colunas que o permitam. O guia da API de execução diferida ilustra a aplicação antecipada de predicados e projeções com uma agregação filtrada.

A projeção mantém as colunas de que uma consulta precisa. A aplicação antecipada de predicados aproxima da origem as condições de seleção de linhas adequadas. Nenhuma destas técnicas permite alterar o que o consumidor recebe. Se um processo a jusante precisar de um identificador, removê-lo altera o contrato de dados. O mesmo acontece ao eliminar valores em falta que tenham significado.

Nem todas as expressões podem ser transferidas para o leitor. CSV e Parquet não têm necessariamente as mesmas reduções de E/S. Inspecione o plano e depois meça a E/S antes de afirmar que houve uma redução.

Porque read().lazy() não pode desfazer uma leitura imediata

Quando read_parquet(...).lazy() chega a .lazy(), o carregamento imediato já aconteceu. As transformações posteriores podem ser diferidas; a leitura inicial não pode ser desfeita. A documentação de origens e destinos recomenda scan_* em vez de read_* para fluxos de trabalho com ficheiros que se pretendam executar de forma diferida.

Isso não significa que um DataFrame carregado imediatamente seja um erro. O guia da API de execução diferida descreve o seu valor durante a exploração, quando os resultados intermédios ajudam a compreender os dados. Escolha-o por esse motivo, em vez de reescrever trabalho exploratório útil apenas para lhe chamar processamento em fluxo.

O exemplo aqui apresentado pressupõe um ficheiro local hipotético events.parquet. As suas colunas record_id e category são do tipo String; value é do tipo Float64. Nenhum esquema foi inspecionado. Os dois possíveis consumidores precisam de registos selecionados ao nível da linha ou de um resumo por categoria derivado desses registos.

Analise separadamente o plano lógico e o plano físico de processamento em fluxo

Use explain para inspecionar a otimização

Compare query.explain(optimized=False) com query.explain(optimized=True). O guia de planos de consulta mostra um filtro a ser transferido para uma leitura diferida após a otimização. Estas vistas descrevem o trabalho planeado. Não medem o tempo de execução nem o pico de memória, e não demonstram a correção dos resultados.

Verifique se a projeção e a seleção esperadas aparecem. Se alguma estiver em falta, investigue antes de interpretar as medições de recursos. Depois, confronte o próprio filtro com a regra de negócio: um filtro no lugar certo pode ainda selecionar as linhas erradas.

Use o grafo físico para investigar operações sensíveis ao uso de memória

O guia de processamento em fluxo documenta query.show_graph(plan_stage="physical", engine="streaming") e uma legenda para operações potencialmente exigentes em memória. A renderização requer que o Graphviz esteja disponível em PATH, como explica o guia de planos de consulta. Sem ele, a versão textual de explain continua a ser útil, mas não oferece a mesma vista física.

O processamento em fluxo funciona por lotes. Algumas operações são, por natureza, incompatíveis com o processamento em fluxo; outras ainda não têm uma implementação para esse modo e recorrem à execução em memória. Use o grafo físico para identificar essas questões. Ele não garante um limite para o pico de RSS.

Verifique também os valores predefinidos específicos da versão. A atual referência estável de collect indica que engine="auto" segue Config.set_engine_affinity ou POLARS_ENGINE_AFFINITY; na ausência destes, usa o processamento em fluxo como predefinição. Instalações mais antigas podem ter um comportamento diferente. Especificar engine="streaming" declara a sua intenção, mas não prova que todas as operações sejam executadas em fluxo. Registe a versão instalada e a configuração antes de se apoiar na documentação atual.

Escolha collect ou sink_parquet com base no resultado de que precisa

Pergunte de que precisa o próximo consumidor. O tamanho do ficheiro de origem, por si só, não permite responder.

Um resultado em memória com limites definidos

collect materializa um LazyFrame num DataFrame. Esse objeto devolvido precisa de memória, mesmo que o trabalho a montante seja executado em fluxo. A agregação, por si só, não o torna pequeno: o número de categorias determina quantas linhas de resumo permanecem.

Um resultado persistido sem recolher toda a saída

A referência de sink_parquet descreve a avaliação em fluxo e a escrita em Parquet sem exigir primeiro que o código que faz a chamada obtenha o DataFrame final completo. A predefinição lazy=False executa a escrita no destino. Com lazy=True, a execução aguarda a recolha; os argumentos engine e optimizations da escrita no destino não se aplicam à construção desse plano diferido. Escolha as definições de execução ao recolhê-lo. A opção diferida está assinalada como instável.

Distinga as regras de recurso a modos alternativos dos dois métodos. A referência atual da escrita no destino segue a afinidade do motor para a seleção automática; na ausência desta, usa o processamento em fluxo, e descreve o recurso ao processamento em fluxo como alternativa ao motor selecionado. A referência de collect descreve o recurso à execução em memória como alternativa ao motor selecionado. As limitações ao nível das operações continuam a aplicar-se. As definições de compressão e de grupos de linhas não estabelecem um orçamento fixo de memória.

A pergunta histórica da comunidade sobre DuckDB e Polars aborda a captura de alterações de dados do PostgreSQL. Leia-a como uma questão de terminologia, não como aconselhamento de arquitetura verificado. O processamento em fluxo por lotes aqui descrito não fornece um serviço permanente com posições de leitura, tratamento do tempo dos eventos ou garantias de entrega.

O Guia Polars orientado para o resultado

Defina o contrato do resultado antes de escolher a execução

O Guia Polars orientado para o resultado é uma lista de verificação editorial original, não uma biblioteca nem uma implementação validada. Comece por identificar o consumidor. Acorde as colunas e os tipos necessários e, depois, defina os filtros, o tratamento de valores nulos e eventuais requisitos de ordenação. Essas decisões dão-lhe algo concreto para validar após a execução.

O diagrama mostra, de forma conceptual, a escolha do destino. Não é um plano físico do Polars observado numa execução.

Leia de forma diferida, restrinja e inspecione

Esta ilustração pressupõe que o Polars já está disponível e usa a API estável atualmente documentada. Não foi executada. Não foram verificadas a configuração nem o conteúdo do conjunto de dados, e não se afirma que tenha sido produzido qualquer resultado impresso. O filtro elimina linhas com value nulo apenas para ilustrar uma política; o seu conjunto de dados pode exigir outra.

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 resolve o esquema planeado. Isto pode ler metadados do disco ou fazer pedidos remotos, pelo que a resolução do esquema não é necessariamente isenta de custos. Também não valida todas as linhas face às regras do domínio. Compare os nomes e os tipos com o contrato e, depois, inspecione o plano lógico e, quando as ferramentas o permitirem, o grafo físico. O guia de esquemas explica porque os esquemas conhecidos são importantes durante o planeamento da execução diferida.

Escolha um dos seguintes métodos de execução para o resultado necessário. São alternativas não executadas, não uma sequência. O ramo do resumo pressupõe que tenha estabelecido, de forma independente, um número de categorias que possa gerir. Também deve decidir o que significa uma categoria nula.

A alternativa A produz o resumo por categoria necessário em Python.

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

A alternativa B escreve os registos selecionados necessários em disco.

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

A API de collect e a API de escrita no destino sustentam estas descrições dos métodos de execução. Comparar os seus tempos seria comparar tarefas diferentes: uma devolve um resumo por categoria, enquanto a outra exporta registos ao nível da linha. Isso não é um teste de desempenho do motor.

Proponha uma verificação comparável e depois valide o destino

Uma comparação útil mantém o resultado constante. Use a mesma versão dos dados de entrada e a mesma versão do Polars, com definições, máquina e condições de armazenamento comparáveis. O guia da Optijara sobre manter os protocolos de testes de desempenho associados aos resultados aborda essa disciplina noutro domínio. Mantenha o protocolo proposto separado dos resultados que vier a registar.

Nenhuma destas medições foi realizada aqui. Para um candidato persistido, verifique o esquema e as contagens de linhas sem carregar toda a saída apenas para a validar. Compare os valores com as tolerâncias acordadas; compare a ordem apenas quando o contrato o exigir.

Se a execução falhar, trate os ficheiros parciais como não validados. Investigue o erro e as operações envolvidas. Use um novo destino candidato na tentativa seguinte e valide-o antes de qualquer entrega aprovada. A presença de um ficheiro não constitui uma verificação de aceitação.

O que as equipas interpretam mal na execução diferida e em fluxo

O atalho tentador é tratar o planeamento da execução diferida e a execução em fluxo como uma única decisão. São decisões separadas. Acrescentar .lazy() depois de uma leitura não pode alterar esse carregamento já concluído. Escolher o processamento em fluxo não elimina o DataFrame final devolvido por collect.

A inspeção também pode parecer mais conclusiva do que é. Um plano explica o trabalho pretendido; um esquema resolvido fornece nomes e tipos. Nenhum deles responde se o uso de memória é aceitável ou se as linhas mantêm o significado pretendido. Mantenha essas verificações separadas da inspeção.

Uma comparação de desempenho torna-se difícil de interpretar se alterar os filtros ou a projeção entre ensaios. Alterar a versão dos dados ou o destino cria o mesmo problema. Registe a alteração da tarefa em vez de atribuir todas as diferenças ao motor. E mantenha os requisitos de ingestão contínua fora da decisão sobre um método de processamento por lotes.

Para linhas extraídas de PDF, verifique se os cabeçalhos e as unidades continuam associados aos valores corretos e mantenha as referências às fontes antes da transformação tabular. O guia da Optijara sobre preservar o significado das tabelas extraídas de PDF e as referências às fontes abrange esse trabalho a montante. Não demonstra uma integração com o Polars. Um destino de escrita diferente não pode recuperar significado perdido durante a extração.

Limitações a verificar antes de alterar um fluxo de processamento de dados

Compromissos entre memória, versão e armazenamento

O tamanho da entrada é apenas uma parte da questão da memória. Considere o estado intermédio e o resultado final e verifique que operações podem ser executadas em fluxo. A documentação de processamento em fluxo e dos métodos de execução explica essas limitações. A adoção também exige esforço de implementação e acrescenta E/S de armazenamento ou complexidade operacional. Não se promete aqui qualquer melhoria de tempo de execução ou de custos.

A investigação de memória da PythonSpeed é um exemplo histórico concreto de análise específica de uma carga de trabalho. As suas medições aplicam-se ao seu conjunto de dados, versões e plataforma. Não são resultados atuais universais nem testes da Optijara.

Antes de usar predefinições ou parâmetros opcionais, consulte a documentação da versão instalada. A referência da escrita no destino assinala várias opções como instáveis ou experimentais. Se quiser compreender uma alteração, evite ajustar a compressão e a ordenação ao mesmo tempo que as definições de execução.

O significado dos dados e a privacidade continuam a precisar de revisão

Durante a preparação, preserve os identificadores e os limites entre partições do conjunto de dados. Torne explícito o tratamento de valores nulos e mantenha a ordenação quando o consumidor a exigir. Um ficheiro Parquet válido não prova que os exemplos pretendidos tenham sido preservados. O guia relacionado sobre preservar a seleção do conjunto de dados e a identidade temporal explica essas verificações para conjuntos de dados dependentes do tempo; não afirma que exista uma integração com o Polars.

Use versões autorizadas dos dados de entrada. Decida quem pode aceder às saídas candidatas e durante quanto tempo devem ser conservadas. Um destino local não resolve as obrigações de privacidade, mesmo que a consulta seja executada com sucesso.

Escolha o destino antes de recolher o resultado

Comece pelo resultado de que o consumidor precisa. Use uma leitura diferida para o trabalho com ficheiros que se pretenda executar de forma diferida e, depois, inspecione os planos lógicos e físicos de acordo com as suas diferentes finalidades. Valide o método de execução escolhido e mantenha provas de medições reais antes de alterar um fluxo operacional. Para ajudar a planear um fluxo de trabalho de IA, a Optijara oferece consultoria em IA.