Otimização Batch com Polars LazyFrame no Cloud Storage

Uma fatura diária de $4.000 no AWS EMR e picos de latência de 45 minutos forçaram nossa equipe a repensar o PySpark para jobs batch de médio porte com a arquitetura de processamento batch polars lazyframe nuvem. Ao lidar com incrementos horários entre 50 GB e 2 TB, o uso de clusters Spark distribuídos introduz custos elevados de garbage collection na JVM, overhead de agendamento de workers e spillage de shuffle desnecessário. Ao executar Polars LazyFrames com processamento via streaming diretamente sobre datasets Parquet no AWS S3, eliminamos o tempo de inicialização de clusters distribuídos e reduzimos os custos de computação significativamente sem falhas de memória nos workers.

O Gargalo de Memória no Processamento de Dados na Nuvem

Arquiteturas de dados modernas frequentemente superdimensionam clusters Spark para workloads que não exigem coordenação multinó complexa. Quando um job batch processa dezenas de gigabytes por hora, o overhead de rede para fazer o shuffle de dados entre nós distribuídos costuma ultrapassar o próprio tempo de processamento computacional. Drivers PySpark ficam limitados por memória ao agregar dados de médio porte, causando erros críticos do tipo java.lang.OutOfMemoryError nos workers.

A transição para contêineres leves rodando engines construídas em Rust oferece uma alternativa estrutural eficiente. O Polars utiliza o formato de memória colunar do Apache Arrow e execução multithreaded em Rust para processar volumes que excedem a RAM do sistema hospedeiro. Em vez de carregar partições inteiras na memória, sua API LazyFrame constrói um plano lógico de execução, otimiza o pushdown de predicados e processa os dados em fluxos contínuos por meio de buffers de memória fixos.

Em nosso pipeline de produção, o processamento de logs transacionais brutos no S3 exigia a leitura de partições Parquet de vários gigabytes, filtragem de schemas inválidos e cálculo de métricas agregadas. No PySpark, esse padrão gerava custos elevados devido ao tempo de inicialização dos executores. Substituindo o PySpark por tarefas conteinerizadas com Polars no AWS Fargate, atingimos menor latência no ciclo completo de ingestão utilizando instâncias menores.

Otimização do Plano Lógico e Pushdown de Predicados

Ao trabalhar com cloud storage como Amazon S3 ou Google Cloud Storage, a largura de banda de rede é o principal gargalo de latência. Ler colunas desnecessárias ou grupos de linhas não filtrados desperdiça IOPS e ciclos de CPU. A interface scan_parquet do Polars resolve isso postergando a execução até que a coleta de dados seja explicitamente solicitada, projetando apenas as colunas necessárias e aplicando filtros diretamente no leitor Parquet.

import polars as pl
import pyarrow.dataset as ds
from pyarrow.fs import S3FileSystem

def execute_cloud_transformation(s3_uri: str, target_output: str):
    s3_fs = S3FileSystem(
        region="us-east-1",
        request_timeout_ms=15000
    )
    
    dataset = ds.dataset(
        s3_uri,
        format="parquet",
        filesystem=s3_fs,
        partitioning="hive"
    )
    
    lazy_plan = (
        pl.scan_pyarrow_dataset(dataset)
        .filter(pl.col("event_status") == pl.lit("COMPLETED"))
        .filter(pl.col("transaction_amount") > 0)
        .with_columns([
            (pl.col("raw_timestamp").str.to_datetime()).alias("event_time"),
            (pl.col("transaction_amount") * pl.col("exchange_rate")).alias("amount_usd")
        ])
        .group_by_dynamic("event_time", every="1h", group_by="merchant_id")
        .agg([
            pl.col("amount_usd").sum().alias("hourly_volume"),
            pl.col("transaction_id").count().alias("transaction_count")
        ])
    )
    
    # Executa em modo streaming para manter uso de memória constante
    sink_query = lazy_plan.collect(streaming=True)
    sink_query.write_parquet(target_output, compression="snappy")

if __name__ == "__main__":
    execute_cloud_transformation(
        s3_uri="s3://production-analytics-lake/raw_transactions/year=2026/month=03/",
        target_output="/tmp/processed_hourly_metrics.parquet"
    )

Garantindo que o pushdown de predicados funcione em conjunto com o leitor S3 do PyArrow, o Polars evita o download de metadados de partições descartadas pela regra de negócio. Esse padrão alinha-se diretamente com arquiteturas modernas de lakehouse, como o projeto AWS Databricks Lakehouse, onde a camada raw exige filtragem de baixo custo antes da gravação nas tabelas silver.

Como o Processamento via Streaming Gerencia Limites de Memória

O mecanismo central que permite ao Polars processar arquivos de 500 GB em uma instância com 32 GB de RAM é sua engine de streaming. Avaliações ansiosas (eager) carregam colunas inteiras em blocos contíguos de memória. Em contrapartida, a execução em streaming avalia o plano lógico em pequenos sub-blocos (morsels) via pipelines paralelos em Rust.

Ao executar .collect(streaming=True), o Polars divide o dataset em lotes físicos dimensionados de acordo com os núcleos de CPU disponíveis. As operações de projeção, filtragem e transformação matemática ocorrem bloco a bloco. Quando a query exige ordenação ou agregação (operadores bloqueantes por natureza), o Polars aplica algoritmos fora de núcleo (out-of-core) otimizados, gravando tabelas hash intermediárias em discos NVMe locais caso o limite de memória seja atingido.

Para gerenciar o impacto de pequenos arquivos gerados em ingestões contínuas, pipelines streaming devem ser pareados com rotinas de manutenção de metadados. Abordagens similares às detalhadas no artigo Iceberg Metadata Compaction garantem que a leitura no cloud storage permaneça de alta performance para engines analíticas downstream.

Benchmark: PySpark vs Polars em Ingestões de Médio Porte

Para comparar a estabilidade operacional e o consumo de recursos, realizamos um benchmark entre PySpark em um cluster EMR e uma tarefa Polars em nó único rodando no AWS ECS Fargate. O volume de dados de teste incluiu 250 GB de arquivos Parquet correspondentes a um dia de logs transacionais.

• Cluster PySpark EMR: 1 Nó Primary (m5.xlarge), 4 Nós Worker (r5.2xlarge). Total alocado: 32 vCPUs, 256 GB RAM.
• Polars ECS Fargate: 1 Contêiner único. Total alocado: 16 vCPUs, 32 GB RAM.

A redução no tempo total decorre da eliminação da serialização entre os ambientes Python e Java (Py4J). Como o Polars opera diretamente sobre os buffers de memória do Arrow sem overhead de IPC entre processos, o uso de CPU é concentrado na execução de instruções vetoriais SIMD em Rust.

Limitações e Considerações Arquiteturais

Embora o Polars LazyFrame ofereça alto rendimento em instâncias únicas, engenheiros de dados precisam considerar as seguintes restrições técnicas ao projetar a infraestrutura:

1. Escalabilidade Horizontal: O Polars não distribui processamento entre múltiplos servidores físicos de forma nativa. Se uma partição individual ou tabela hash intermediária exceder os limites combinados de disco local e RAM em operações não suportadas por streaming, engines distribuídas como PySpark continuam necessárias.
2. Listagem de Objetos no S3: Ler milhões de arquivos pequenos em buckets de armazenamento gera gargalos de requisições HTTP. É fundamental adotar estruturas de partição no padrão Hive ou catálogos de metadados para permitir a descoberta eficiente dos arquivos pelo Polars.
3. Evolução de Schema Strict: Ao contrário do Spark SQL, que realiza coerção implícita de tipos entre arquivos Parquet heterogêneos, o Polars exige que a unificação dos tipos de dados seja feita de forma explícita antes da execução do plano lógico.

Para pipelines que processam volumes diários abaixo da escala de terabytes, a substituição de clusters distribuídos por jobs Polars LazyFrame oferece uma redução substancial de custos de infraestrutura, simplifica os processos de CI/CD e garante previsibilidade de consumo de recursos.