DEV Community: Richardson

Estacionariedade: Por que a Média Histórica é Perigosa para suas Projeções

Richardson — Sun, 04 Jan 2026 21:18:02 +0000

Resumo: Neste artigo, exploramos o conceito de estacionariedade em séries temporais, como utilizar o teste Augmented Dickey-Fuller (ADF) para diagnosticar tendências estocásticas e como configurar o parâmetro de integração (d) no SARIMAX para evitar o viés da média global.

Introdução

No mundo dos negócios, existe uma atração pela "média". Quando um time de stakeholders pede uma projeção para 2026, o instinto inicial de muitos analistas é calcular a média de 2024/2025 e projetá-la à frente.

Se você trabalha com séries temporais financeiras ou de e-commerce, essa abordagem geralmente está errada.

O motivo é a Estacionariedade. Se uma métrica possui uma tendência forte (seja de alta ou queda), sua média e variância não são constantes ao longo do tempo. Projetar uma média global em uma série com tendência resulta em erros grosseiros de previsão.

Neste post, vou mostrar como diagnosticamos esse problema em dados de e-commerce usando o teste Augmented Dickey-Fuller (ADF) e como corrigimos isso matematicamente no modelo SARIMAX.

O Conceito: Raiz Unitária e Estacionariedade

Para um modelo preditivo funcionar bem (especialmente da família ARIMA), a série precisa ser, idealmente, estacionária.

O que é Estacionariedade?
Uma série é estacionária quando suas propriedades estatísticas — média, variância e autocorrelação — são constantes ao longo do tempo. Ela oscila em torno de um valor fixo.

O Problema da Raiz Unitária
Muitas métricas de negócio (como Receita ou Preço) possuem o que chamamos de "Raiz Unitária". Simplificando, isso significa que a série tem uma memória forte: o valor de hoje depende fortemente do valor de ontem, mais um choque aleatório. Isso cria uma tendência estocástica que faz a série "driftar" (derivar) para longe da média histórica.

Se você tentar forçar uma média fixa em uma série com raiz unitária, seu modelo estará sempre "correndo atrás" da tendência, errando sistematicamente.

O Diagnóstico: Teste Augmented Dickey-Fuller (ADF)

Não confiamos apenas no "olhômetro". Para validar se uma série é estacionária, usamos o teste ADF.

Como interpretar o teste:

Hipótese Nula ( $H_0$ ): A série tem uma raiz unitária (NÃO é estacionária).
Hipótese Alternativa ( $H_1$ ): A série é estacionária.
Regra de Decisão: Se o p-valor < 0.05, rejeitamos $H_0$ (a série é estacionária). Caso contrário, ela tem tendência e precisa de tratamento.

Exemplo Prático em Python

Em um estudo recente, rodamos o ADF em diversas métricas usando a biblioteca statsmodels. Veja o código e os resultados reais:

from statsmodels.tsa.stattools import adfuller

def teste_estacionariedade(serie, nome):
    resultado = adfuller(serie.dropna())
    print(f"Métrica: {nome}")
    print(f"Estatística ADF: {resultado}")
    print(f"p-valor: {resultado}")
    if resultado < 0.05:
        print("Resultado: Estacionária (d=0)")
    else:
        print("Resultado: NÃO Estacionária (Requer d=1)")
    print("-" * 30)

# Exemplo de uso com dados reais do nosso dataset
# df_main é o dataframe contendo as séries temporais
teste_estacionariedade(df_main['ticket_medio'], 'Ticket Médio')
teste_estacionariedade(df_main['sessoes'], 'Sessões')

Resultados Obtidos

Ao analisar os resultados e gráficos, encontramos dois cenários que invalidariam qualquer modelo baseado em média simples:

Ticket Médio (Preço):
- p-valor: 0.76
- Diagnóstico: Extremamente não estacionário. A série tem uma tendência de alta clara e forte. O modelo não pode aprender o valor absoluto (ex.: R$ 1.500), pois amanhã ele será maior.

Estatística ADF: -0.9714409080154276
p-valor: 0.7635375639207811
A série parece ser NÃO-ESTACIONÁRIA (p-valor >= 0.05). Falha ao rejeitar H0.
Indício de presença de Tendência.

Sessões (Tráfego):
- p-valor: 0.10
- Diagnóstico: Não estacionário (0.10 > 0.05). Há uma tendência de queda no tráfego. Uma média histórica superestimaria o tráfego futuro.

Estatística ADF: -2.552426454254865
p-valor: 0.10323368222873064
A série parece ser NÃO-ESTACIONÁRIA (p-valor >= 0.05). Falha ao rejeitar H0.
Indício de presença de Tendência.

A Solução: Diferenciação no SARIMAX

Uma vez identificado que $p-valor>0.05p\text{-valor} > 0.05$ , a correção técnica obrigatória é a Diferenciação (Integration).

Em vez de modelar o valor absoluto $Y_t$ , modelamos a diferença entre o valor atual e o anterior:
$ΔYt=Yt−Yt−1\Delta Y_t = Y_t - Y_{t-1}$

Isso transforma uma série com tendência em uma série estacionária de "variações".

Implementação no SARIMAX

No Python, você não precisa criar a coluna de diferença manualmente. O algoritmo SARIMAX possui o parâmetro order=(p,d,q), onde d é a ordem de integração.

Baseados nos testes ADF acima, definimos a estratégia de modelagem:

from statsmodels.tsa.statespace.sarimax import SARIMAX

# Para Ticket Médio (Não-Estacionário, p-valor=0.76)
# Usamos d=1. O modelo aprende a taxa de crescimento, não o valor fixo.
model_ticket = SARIMAX(
    df_main['ticket_medio'],
    order=(1, 1, 1),  # (AR, Integração, MA)
    seasonal_order=(1, 1, 1, 52) # Sazonalidade anual
)

# Para Transações (Estacionário mas com queda estrutural)
# Mesmo que o ADF indique estacionariedade (p < 0.05),
# forçamos d=1 para capturar a 'inércia' da queda recente,
# evitando reversão à média antiga.
model_transacoes = SARIMAX(
    df_main['transacoes'],
    order=(1, 1, 1),
    seasonal_order=(1, 1, 1, 52)
)

model_fit = model_ticket.fit()
forecast = model_fit.get_forecast(steps=52)

O impacto no negócio:
Ao usar d=1, o modelo entendeu que o Ticket Médio estava subindo cerca de R$ 10,00 por semana e projetou essa continuidade para 2026. Se tivéssemos usado d=0 (ou uma média simples), a projeção teria achatado a curva, subestimando a receita futura e ignorando a inflação de preços interna da loja.

Conclusão

Não confie cegamente na média histórica. Séries temporais de negócios raramente ficam paradas.

Rode o teste ADF: É a maneira estatisticamente robusta de verificar se sua métrica está "presa" a uma média ou se está derivando.
Olhe o p-valor: Se $> 0.05$ , sua série não é estacionária.
Use Diferenciação ( $d = 1$ ): Configure seu modelo ARIMA/SARIMAX para projetar a variação, capturando a tendência real.

Fazer ciência de dados aplicada é garantir que a matemática do modelo reflita a realidade econômica do negócio. Ignorar a não-estacionariedade é planejar o futuro olhando para uma foto estática do passado.

Referências e Bibliografia

Introduction to Time Series and Forecasting (Brockwell & Davis) - Para fundamentos matemáticos de estacionariedade.
Documentação Statsmodels (Augmented Dickey-Fuller).
Practical Statistics for Data Scientists (Bruce & Bruce) - Para aplicação prática de conceitos estatísticos em DS.

Regressão Linear para Inferência Causal: Indo Além da Predição

Richardson — Tue, 30 Dec 2025 17:44:11 +0000

Na engenharia de dados, o foco costuma residir na integridade e velocidade do pipeline. No entanto, ao transitar para a modelagem, é fundamental compreender que a Regressão Linear — frequentemente reduzida à fórmula — possui aplicações que extrapolam a simples previsão de valores. Enquanto o aprendizado de máquina convencional prioriza a predição (), a regressão é uma ferramenta de inferência causal, capaz de isolar o impacto de variáveis específicas () sobre um resultado, controlando o ruído estatístico.

1. Diferença entre Predição e Inferência

Em modelos de predição, o objetivo é minimizar o erro (como o RMSE) entre o valor real e a estimativa . O modelo pode ser uma "caixa preta", desde que a acurácia seja alta.

Na Inferência Causal, o interesse reside nos coeficientes de regressão (). O objetivo é mensurar como se altera quando uma variável é modificada, mantendo todos os outros fatores constantes.

2. Estudo de Caso: Qualidade de Cadastro no E-commerce

Recentemente fiz um estudo sobre o impacto da qualidade do cadastro na conversão de vendas que ilustra essa aplicação. Em vez de apenas prever vendas, a regressão foi utilizada como diagnóstico de negócio.

Discretização (Variáveis Dummy): A análise exploratória revelou que o ganho de conversão não era linear, mas ocorria em "degraus". Foi criada uma variável binária is_score_elite, onde 1 representa score 80 e 0 para valores menores.

Controle de Variáveis: Para evitar que o efeito do preço ou do frete mascarasse o impacto da qualidade, utilizou-se a Regressão Múltipla:
Resultados e Métricas: O modelo apresentou um de apenas 0,024 (2,4%). Para predição, esse valor seria insuficiente, mas para inferência, ele foi aceitável, pois o objetivo era isolar o coeficiente . O resultado indicou um ganho de conversão de 10,3% atribuído puramente à qualidade do cadastro.

3. Outras Aplicações Técnicas

A regressão atua como um "bisturi estatístico" em diversos domínios onde experimentos controlados (Testes A/B) são difíceis ou impossíveis:

Marketing e Influência Social: A técnica permite distinguir homofilia (conexão entre pessoas similares) de influência real. Ao controlar características demográficas e gostos prévios, a regressão revela se uma compra ocorreu devido à influência de um terceiro ou apenas por afinidade pré-existente entre os indivíduos.
Setor Imobiliário (Precificação Hedônica): Para determinar o valor causal de um cômodo adicional, a regressão múltipla isola variáveis de confusão, como a localização. Sem esse controle, modelos simples podem indicar erroneamente que mais quartos diminuem o valor da casa, apenas porque casas maiores em áreas rurais são mais baratas.
Saúde Pública: Quando não é ético realizar experimentos, a regressão múltipla fixa variáveis como idade e histórico médico para identificar o impacto isolado de um medicamento ou hábito sobre a saúde, mitigando fenômenos como o Paradoxo de Simpson.

4. Limitações e Boas Práticas

Para garantir a validade da inferência, o engenheiro de dados deve estar atento a:

Variáveis Omitidas: A ausência de uma variável importante pode inflar artificialmente os coeficientes de outras variáveis, gerando viés.
Correlação vs. Causalidade: A matemática identifica associações; o conhecimento de domínio é necessário para interpretar a causalidade.
Extrapolação: Modelos de regressão perdem a validade quando aplicados a intervalos de dados fora do conjunto de treinamento.

Referências Bibliográficas

PROVOST, Foster; FAWCETT, Tom. Data Science for Business.
SCHUTT, Rachel; O'NEIL, Cathy. Doing Data Science.
BRUCE, Peter; BRUCE, Andrew. Practical Statistics for Data Scientists.

Guia arquitetônico de ponta para a construção de uma plataforma de dados

Richardson — Sun, 12 Oct 2025 03:16:17 +0000

Etapa 1: A Conexão Fundamental - Modelo Lógico/Físico e a Arquitetura Medalhão

A Arquitetura Medalhão é a estrutura que nos permite aplicar a modelagem de dados de forma estratégica, conectando o caos da origem à clareza do consumo. A separação entre o modelo lógico e o físico se manifesta de forma clara no fluxo entre as camadas.

Camada Bronze para Prata (Foco na Criação do Modelo Lógico Normalizado): A transformação fundamental aqui é a imposição de sentido, integridade e governança. Pegamos dados brutos (um modelo físico simples, mas logicamente caótico) e os forjamos em um modelo lógico e normalizado que representa as entidades e processos de negócio de forma clara e consistente. Metodologias baseadas em Modelagem de Entidade-Relacionamento (ER) ou Terceira Forma Normal (3FN) são ideais para essa camada de integração, visando a verdade semântica, a integridade e a redução da redundância de dados.
Camada Prata para Ouro (Foco na Otimização do Modelo Físico Desnormalizado): Com a verdade lógica já estabelecida na camada Prata, o foco muda completamente para a entrega de performance e simplicidade para o usuário final. Pegamos o modelo lógico normalizado (que é ótimo para integridade, mas ruim para performance de BI devido aos JOINs) e criamos uma representação física otimizada para casos de uso específicos. Aqui aplicamos estratégias de desnormalização, cuja manifestação mais comum é a Modelagem Dimensional (Star Schema) ou, em sua forma mais extrema, a One Big Table (OBT). O sucesso dessa estratégia é garantido pela natureza do armazenamento colunar do BigQuery, que assegura que as consultas leiam apenas os dados relevantes, superando a ineficiência dos JOINs para o consumidor final.

Etapa 2: A Estrutura Detalhada das Camadas no Google Cloud

Apresento a estrutura detalhada revisada, incorporando as melhores práticas e as nuances discutidas.

Camada Bronze: O Data Lake Bruto e Imutável

Categoria	Detalhes e Conceitos
Propósito Principal	Ingestão e persistência de dados brutos, imutáveis e históricos. É o "backup" auditável da realidade da fonte.
Abstração Envolvida	"Aterrissagem de Dados" (Data Landing Zone). É um repositório que aceita dados em qualquer formato e velocidade.
Estrutura (Modelo)	Schema-on-Read O modelo físico é simples: uma linha por registro. O modelo lógico é indefinido.
Metodologias e Padrões	• Ingestão Imutável: Nunca alterar um dado na Bronze.• Ingestão Desacoplada (CDC/Streaming): Prioriza a replicação de logs (via Datastream) ou a captura de eventos (via Pub/Sub) para minimizar o impacto nos sistemas de origem (OLTP). • Formatos de Arquivo Otimizados: Priorizar formatos colunares como Apache Parquet ou baseados em esquema como Apache Avro.
Design Partners	• Engenheiros de Dados: Construtores dos pipelines.• Auditores e Equipes de Conformidade: Utilizam a Bronze para rastrear a linhagem.
Tecnologias Google Cloud	• Cloud Storage (GCS): O repositório primário e ideal para o Data Lake Bruto.• BigQuery: Atua como componente de apoio (sink para streaming/CDC ou motor de consulta via tabelas externas).• Dataplex: Para catalogação de dados, descoberta de metadados e governança centralizada desde a ingestão. • Pub/Sub, Datastream: Serviços de ingestão (o "E" e "L" do ELT).• Cloud Composer (Airflow): Orquestrador principal para agendar e gerenciar o fluxo de ingestão.
Desafios Comuns	• Schema Drift: A estrutura dos dados na fonte muda.• Governança (Data Swamp): Risco de se tornar um "pântano de dados". Mitigado pela catalogação proativa com Dataplex para garantir linhagem, documentação e detecção de PII (via integração com DLP).• Gerenciamento de Custos: Implementar políticas de ciclo de vida (Lifecycle Management) no GCS para mover dados para classes de armazenamento mais frias.
Exemplo Prático	Arquivos Avro armazenados no GCS em `gs://ecommerce-bronze/...`, com metadados gerenciados pelo Dataplex.

Camada Prata: A Fonte da Verdade Normalizada e Confiável

Categoria	Detalhes e Conceitos
Propósito Principal	Limpar, validar, enriquecer e integrar os dados brutos. É a "Fonte Única da Verdade" (SSOT) e o local de implementação da governança de dados de baseline.
Abstração Envolvida	"Hub de Integração" (Integration Hub) Aqui criamos um modelo de dados corporativo consistente.
Estrutura (Modelo)	Modelo Lógico Normalizado (3FN, Modelo ER). A prioridade é a integridade. O modelo físico é otimizado com Particionamento e Clustering para acelerar as operações de integração.
Metodologias e Padrões	• Modelagem de Entidade-Relacionamento (ER) / 3FN: Para garantir a integridade.• Data Vault: Metodologia robusta para ambientes com alta variação de esquema.• Regras de Qualidade de Dados (DQ): Implementação de testes automatizados.
Design Partners	• Engenheiros de Dados: Construtores do modelo.• Analistas e Cientistas de Dados: Consumidores para explorações profundas.
Tecnologias Google Cloud	• BigQuery: O coração da camada Prata, executando as transformações.• Dataform / dbt: Ferramentas para orquestrar as transformações SQL-first e injetar testes de validação (DQ).• Cloud Composer (Airflow): Orquestra o pipeline ponta-a-ponta, acionando os jobs do Dataform/dbt.
Desafios Comuns	• Lógica de Negócio Complexa: Traduzir regras de negócio em um modelo normalizado.• Custo e Complexidade dos `JOINs`: A normalização exige `JOINs` computacionalmente caros.• Manutenção do Modelo: Atualizar o modelo à medida que o negócio evolui.
Exemplo Prático	Um conjunto de tabelas normalizadas em `ecommerce_silver`: `clientes`, `produtos`, `pedidos`, etc.

Camada Ouro: Os Produtos de Dados Focados no Negócio

Categoria	Detalhes e Conceitos
Propósito Principal	Fornecer dados prontos para consumo, agregados, desnormalizados e otimizados para casos de uso específicos com máxima performance.
Abstração Envolvida	"Produto de Dados" (Data Product) Cada tabela na Ouro é um produto curado, que inclui dados, metadados, governança e testes, tornando-o self-service e confiável.
Estrutura (Modelo)	Modelo Físico Desnormalizado e Otimizado. O objetivo é eliminar `JOINs` em tempo de consulta. A estrutura utiliza Particionamento e Clustering e recursos nativos como STRUCTs e ARRAYs.
Metodologias e Padrões	• Modelagem Dimensional (Star Schema): Padrão para data marts de BI. • Criação de Tabelas Amplas (OBT): Para dashboards de alta performance.• MLOps (Feature Engineering): Orquestração do ciclo de vida de modelos de ML, desde a featurização até o treinamento e a predição.
Design Partners	• Analistas de Negócio, Executivos (via dashboards): Consumidores finais.• Aplicações e APIs: Consomem dados da camada Ouro.• Engenheiros de ML: Utilizam o Ouro para MLOps.
Tecnologias Google Cloud	• BigQuery: O motor de serviço serverless perfeito. • Looker / Looker Studio: Ferramentas de BI que se conectam à camada Ouro.• Vertex AI (incluindo Pipelines): Consome tabelas da Ouro para MLOps. Vertex AI Pipelines orquestra o ciclo de vida do ML.• Cloud Composer (Airflow): Orquestra a execução dos jobs que atualizam esta camada.
Desafios Comuns	• Explosão de Marts: Criar dezenas de tabelas Ouro sem governança, levando à inconsistência.• Balanceamento da Granularidade: Decidir a agregação correta.• Custo Computacional Alto: A construção da camada Ouro envolve `JOINs` e agregações custosas.
Exemplo Prático	• `ecommerce_gold.dm_vendas`: Um Star Schema para análise de BI. • `ml_gold.customer_features`: Uma tabela de features para um modelo de churn.

Etapa 3: Práticas Transversais Essenciais

Além da estrutura em camadas, práticas de engenharia de ponta são cruciais para o sucesso da plataforma.

Governança Proativa com Catálogo de Dados: Para mitigar o risco de "pântano de dados", a integração com o Dataplex deve ser implementada desde a camada Bronze. Isso garante que metadados, linhagem de dados e detecção de PII (via integração com o Cloud DLP) sejam implementados desde o início, aumentando a governança e a capacidade de descoberta de todos os ativos de dados.
Gestão Estratégica de Custos no BigQuery: O controle de custos vai além do design das tabelas. É fundamental utilizar features específicas do BigQuery para otimizar o processamento, como a alocação de capacidade com BigQuery Editions/Reservations para cargas de trabalho previsíveis e o uso de dry-run (simulação) para estimar o volume de dados lidos antes da execução de consultas complexas.

Etapa 4: A Síntese Revisada - O Fluxo de Valor Orquestrado

A jornada de um dado através desta arquitetura representa um fluxo contínuo de agregação de valor, orquestrado de ponta a ponta:

Um pipeline de dados, orquestrado pelo Cloud Composer, é iniciado. Ele usa o Datastream para capturar mudanças e aterrissá-las como arquivos Avro no Cloud Storage (Camada Bronze), com seus metadados sendo registrados no Dataplex.
O Composer aciona um job do Dataform. Este job lê os arquivos da Bronze, executa SQLs que limpam, validam, aplicam testes de qualidade (DQ) e inserem os dados em um modelo normalizado (3FN) na camada Prata do BigQuery.
Com a conclusão da Prata, o Composer aciona um segundo job do Dataform que lê as tabelas normalizadas, executa os JOINs e agregações necessários para desnormalizar os dados, e constrói os "Produtos de Dados" (ex.: um Star Schema) na camada Ouro.
Em paralelo, o Composer pode acionar um pipeline do Vertex AI, que consome uma tabela da camada Ouro para retreinar um modelo de ML, orquestrando todo o ciclo de vida do MLOps.
Finalmente, o Looker se conecta aos Produtos de Dados da camada Ouro, permitindo que os usuários de negócio explorem informações atualizadas e confiáveis com altíssima performance.

Otimizando Redshift na Prática: Um Estudo de Caso com DISTKEY e SORTKEY

Richardson — Wed, 08 Oct 2025 01:07:34 +0000

Neste guia, vamos otimizar uma tabela no Amazon Redshift do zero. Analisaremos o padrão de consulta, usaremos a view svv_table_info para diagnosticar problemas e aplicaremos as DISTKEY e SORTKEY corretas para transformar a performance.

O Amazon Redshift é um data warehouse com arquitetura MPP (Massively Parallel Processing). Em resumo, dados e processamento são divididos entre múltiplos nós. Para extrair a performance máxima, precisamos instruir o Redshift sobre como distribuir (DISTKEY) e ordenar (SORTKEY) os dados de forma inteligente.

Embora o Redshift ofereça otimizações automáticas (AUTO), elas são um ponto de partida genérico. A otimização manual, baseada em padrões de consulta conhecidos, é o que realmente faz a diferença.

Parte 1: Conceitos Fundamentais

Antes de otimizar, precisamos dominar duas ferramentas.

A. Chaves de Distribuição (DISTKEY)

A DISTKEY define como as linhas de uma tabela são distribuídas entre os nós do cluster. O objetivo é minimizar a movimentação de dados pela rede (data shuffling) durante a execução de uma query.

DISTSTYLE KEY(coluna): Linhas com o mesmo valor na coluna da DISTKEY são armazenadas no mesmo nó. Ideal para colunas com alta cardinalidade usadas em JOINs ou GROUP BY, pois a operação ocorre localmente em cada nó.
DISTSTYLE ALL: Uma cópia completa da tabela é armazenada em cada nó. Use apenas para tabelas pequenas e de baixa frequência de atualização (ex: tabelas de dimensão < 3 milhões de linhas) que são frequentemente usadas em JOINs.
DISTSTYLE EVEN: Os dados são distribuídos em round-robin. É o padrão, mas raramente a melhor escolha para tabelas grandes, pois não otimiza a co-localização de dados para JOINs.

B. Chaves de Ordenação (SORTKEY)

A SORTKEY define a ordem física em que as linhas são armazenadas nos blocos de 1MB em disco. O objetivo é minimizar a quantidade de dados lidos (I/O).

O Redshift mantém metadados (Zone Maps) que registram os valores mínimo e máximo de cada bloco. Se uma query filtra por uma coluna da SORTKEY (ex: WHERE data > '2025-10-01'), o Redshift usa os Zone Maps para ler apenas os blocos que podem conter esses dados, ignorando o resto.

COMPOUND SORTKEY(col1, col2, ...): Ordena os dados estritamente na ordem das colunas listadas. É extremamente eficiente quando os filtros usam um prefixo da chave (principalmente a col1).
INTERLEAVED SORTKEY(col1, col2, ...): Dá peso igual a todas as colunas na chave. É útil quando os filtros são imprevisíveis, mas tem um custo maior de VACUUM.

Parte 2: A Ferramenta de Diagnóstico: `svv_table_info`

Esta view do sistema é o painel de saúde de qualquer tabela. Para usá-la, rode:

SELECT "table", diststyle, skew_rows, sortkey1, unsorted, stats_off
FROM svv_table_info
WHERE "schema" = 'seu_schema' AND "table" = 'sua_tabela';

Abaixo, os campos essenciais e como interpretá-los:

Campo Relevante	O que é?	Como Interpretar e Agir?
`diststyle`	O estilo de distribuição atual.	Confirma se sua `DISTKEY` foi aplicada. Se for `EVEN` em uma tabela grande, é um forte candidato à otimização.
`skew_rows`	A distorção na distribuição. Razão entre o maior e o médio slice.	`1.0` é perfeito. Valores altos (`> 4.0`) indicam "skew", onde um nó está sobrecarregado. A `DISTKEY` pode estar mal escolhida.
`sortkey1`	A primeira coluna da sua `SORTKEY`.	Essencial para garantir que a coluna mais filtrada em ranges (como datas) esteja aqui.
`unsorted`	O percentual de dados não ordenado.	O objetivo é `0.00`. Um valor alto anula os benefícios da `SORTKEY`. Ação: `VACUUM SORT sua_tabela;`
`stats_off`	O percentual de "desatualização" das estatísticas.	O objetivo é `0.00`. Um valor alto pode gerar planos de execução ineficientes. Ação: `ANALYZE sua_tabela;`

Parte 3: O Ciclo de Otimização - Estudo de Caso

Vamos aplicar os conceitos à tabela agg_cotacoes_produto_fornecedor_dia.

Etapa 1: Análise do Padrão de Consulta

Problema: A tabela é a fonte de dashboards no Looker, e as consultas estão lentas.
Padrão Identificado: Os filtros dos dashboards são quase sempre por id_fornecedor e por um range de data_cotacao.

Etapa 2: Diagnóstico da Tabela Original

Uma consulta na svv_table_info revelou:

diststyle: AUTO(KEY(id_produto))
sortkey1: AUTO(SORTKEY)

Conclusão do Diagnóstico: A otimização automática do Redshift escolheu id_produto, provavelmente para otimizar JOINs com uma tabela de produtos. No entanto, essa escolha não atende ao nosso padrão principal de filtro (id_fornecedor e data_cotacao).

Etapa 3: Definição da Estratégia de Otimização

Baseado no padrão de consulta, definimos a nova estratégia:

DISTKEY(id_fornecedor): Como os filtros são por fornecedor, isso vai co-localizar os dados necessários em um mesmo nó, acelerando filtros e agregações.
COMPOUND SORTKEY(data_cotacao, id_fornecedor): data_cotacao é a coluna mais filtrada em ranges, então deve vir primeiro na SORTKEY. Isso permitirá ao Redshift pular blocos de dados massivamente.

Etapa 4: Implementação (Processo "Deep Copy")

Para aplicar as mudanças em uma tabela populada, o processo mais seguro é criar uma nova tabela otimizada e depois trocá-las de lugar.

1. Criação (CTAS):

CREATE TABLE agg_cotacoes_produto_fornecedor_dia_novo 
DISTKEY(id_fornecedor) 
COMPOUND SORTKEY(data_cotacao, id_fornecedor) AS 
SELECT * FROM agg_cotacoes_produto_fornecedor_dia;

2. Manutenção Pós-Criação:
A svv_table_info da nova tabela mostrou unsorted e stats_off altos, o que é esperado após uma carga massiva.

VACUUM SORT ONLY agg_cotacoes_produto_fornecedor_dia_novo;
ANALYZE agg_cotacoes_produto_fornecedor_dia_novo;

3. Validação Final:
Verificamos a svv_table_info novamente. Agora, unsorted: 0.00 e stats_off: 0.00. A tabela está pronta.

4. A Troca Atômica:
Executamos a troca de nomes dentro de uma transação para garantir que a operação seja instantânea e segura, sem downtime para os usuários.

BEGIN;
ALTER TABLE agg_cotacoes_produto_fornecedor_dia RENAME TO agg_cotacoes_produto_fornecedor_dia_old;
ALTER TABLE agg_cotacoes_produto_fornecedor_dia_novo RENAME TO agg_cotacoes_produto_fornecedor_dia;
COMMIT;

Conclusão

O ciclo de otimização no Redshift é contínuo e baseado em dados:

Analisar Padrão de Query
Diagnosticar com svv_table_info
Definir Estratégia (DISTKEY, SORTKEY)
Implementar e Validar

Ao dominar esses conceitos, você sai do modo "automático" e passa a ter controle total sobre a performance do seu data warehouse, transformando tabelas lentas em fontes de dados eficientes para qualquer ferramenta de BI como o Looker.

Modelagem de Dados para Análise de Funis no Amazon Redshift

Richardson — Sat, 27 Sep 2025 01:43:59 +0000

Como engenheiros de dados, frequentemente enfrentamos o desafio de transformar logs transacionais—sequências longas de eventos—em uma visão consolidada que permita análises de negócio. Um caso clássico é a modelagem de um funil de conversão, como o rastreamento do ciclo de vida de tickets de suporte em um sistema de atendimento ao cliente.

Neste post, vamos explorar as decisões de arquitetura e as técnicas de SQL para construir uma Tabela Analítica eficiente para esse cenário no Amazon Redshift, comparando duas abordagens de modelagem (One Big Table vs. Star Schema) e três técnicas de pivotagem de dados.

O Cenário: Funil de Tickets de Suporte

Imagine uma tabela de logs, suporte_ticket_eventos, com milhões de linhas e a seguinte estrutura:

id_evento	id_ticket	status_novo	timestamp
901	123	Aberto	2025-09-20 10:00
902	124	Aberto	2025-09-20 10:05
905	123	Em Análise	2025-09-20 11:30
908	123	Resolvido	2025-09-21 14:00
909	124	Resolvido	2025-09-20 18:45

O objetivo de negócio é responder a perguntas como: "Qual o tempo médio entre a abertura e a resolução de um ticket?". Para isso, precisamos de uma tabela final com uma linha por ticket, contendo as datas de cada marco importante.

Decisão 1: A Arquitetura da Tabela - OBT vs. Star Schema

A primeira grande decisão é como estruturar nossa tabela analítica final.

Característica	One Big Table (OBT) / Desnormalizada	Star Schema (Esquema Estrela)
Estrutura	Tabela única e larga com `id_ticket`, `data_abertura`, `data_analise`, `data_resolucao`, `nome_cliente`, `categoria_produto`, etc.	Tabela Fato (`fato_tickets`) com `id_ticket`, `tempo_resolucao_horas` e chaves para as Dimensões `dim_cliente` e `dim_produto`.
Performance de Leitura	Excelente. Sem `JOINs`, as consultas para dashboards são instantâneas.	Muito boa. `JOINs` otimizados entre a tabela fato e as dimensões.
Flexibilidade	Menor. Otimizada para o processo de funil. Perguntas sobre novas dimensões exigem alterações no ETL.	Excelente. Permite análises exploratórias (ad-hoc), cruzando fatos com qualquer combinação de dimensões.
Manutenção	Lógica complexa e centralizada no ETL de criação da tabela.	Mais simples. Atualizar o nome de um cliente, por exemplo, afeta apenas a `dim_cliente`.
Ideal Para	Dashboards de alta performance e análise de processos específicos como funis.	Business Intelligence geral e exploração de dados, criando uma "fonte da verdade" reutilizável.

Por que a OBT é frequentemente a escolha certa para funis?
Para analisar um processo sequencial como um funil, a OBT é superior. A complexidade de calcular os tempos entre múltiplos estados (data_resolucao - data_abertura) é resolvida uma única vez durante o ETL. A estrutura é desenhada especificamente para responder a perguntas sobre esse funil da forma mais rápida possível. Apresentar uma única linha por ticket com todos os seus marcos temporais é a forma mais intuitiva e performática para o consumo em dashboards.

Decisão 2: A Técnica de Pivotagem no Redshift

Uma vez decidida a estrutura OBT, precisamos transformar nossas linhas de log em colunas. Existem várias formas de fazer isso em SQL.

Abordagem 1: Agregação Condicional (Recomendada)

Esta técnica utiliza GROUP BY e CASE WHEN dentro de funções de agregação. É a abordagem mais idiomática e performática em bancos de dados colunares como o Redshift.

CREATE TABLE tickets_funil AS
SELECT
    id_ticket,
    MIN(CASE WHEN status_novo = 'Aberto'      THEN "timestamp" END) AS data_abertura,
    MIN(CASE WHEN status_novo = 'Em Análise'  THEN "timestamp" END) AS data_inicio_analise,
    MAX(CASE WHEN status_novo = 'Resolvido'   THEN "timestamp" END) AS data_resolucao
FROM
    suporte_ticket_eventos
GROUP BY
    id_ticket;

Por que funciona tão bem no Redshift?

Single Pass: A tabela de logs é lida apenas uma vez.
Execução Colunar: O Redshift lê apenas as colunas necessárias (id_ticket, status_novo, timestamp), ignorando o resto e minimizando I/O.
Processamento Paralelo (MPP): A operação GROUP BY é massivamente paralelizada entre os nós do cluster.

Abordagem 2: Múltiplos Self-Joins (Não Recomendada)

Uma abordagem intuitiva para quem vem de bancos de dados transacionais é criar uma subquery para cada status e juntá-las com LEFT JOIN.

-- NÃO FAÇA ISSO EM REDSHIFT PARA PIVOTAGEM
SELECT
    base.id_ticket,
    abertos.data_abertura,
    resolvidos.data_resolucao
FROM
    (SELECT DISTINCT id_ticket FROM suporte_ticket_eventos) AS base
LEFT JOIN
    (SELECT id_ticket, MIN("timestamp") AS data_abertura FROM suporte_ticket_eventos WHERE status_novo = 'Aberto' GROUP BY 1) AS abertos
    ON base.id_ticket = abertos.id_ticket
LEFT JOIN
    (SELECT id_ticket, MAX("timestamp") AS data_resolucao FROM suporte_ticket_eventos WHERE status_novo = 'Resolvido' GROUP BY 1) AS resolvidos
    ON base.id_ticket = resolvidos.id_ticket;

Por que esta abordagem é ruim? Ela força o Redshift a escanear a tabela suporte_ticket_eventos múltiplas vezes, uma para cada subquery. O plano de execução se torna muito mais caro, com mais I/O e movimentação de dados entre os nós.

Alternativa: Funções de Janela (`Window Functions`)

Funções de janela como ROW_NUMBER() ou LEAD()/LAG() são extremamente poderosas, mas para o problema simples de pivotagem, a agregação condicional é geralmente mais direta e performática. As funções de janela brilham em análises de sequência mais complexas, como "qual foi o tempo gasto no status anterior, qualquer que seja ele?".

Dicas Finais de Otimização no Redshift

Para garantir que a abordagem de agregação condicional seja ainda mais rápida, otimize sua tabela de logs:

DISTKEY (Chave de Distribuição): Defina a DISTKEY como o campo de agrupamento (ex: id_ticket). Isso garante que todos os eventos do mesmo ticket fiquem no mesmo nó, eliminando a movimentação de dados pela rede (shuffle) durante o GROUP BY.
SORTKEY (Chave de Ordenação): Use uma chave de ordenação composta (ex: id_ticket, timestamp). Isso acelera a busca e a ordenação dos eventos de um mesmo ticket.

A documentação da AWS sobre "Amazon Redshift Engineering’s Advanced Table Design Playbook: Distribution Styles and Distribution Keys" oferece excelentes guias sobre estes conceitos.

Conclusão

A modelagem de dados para análise de funis exige decisões de arquitetura deliberadas. Para cenários focados em processos, como o nosso exemplo de tickets, a abordagem One Big Table oferece performance e clareza para o usuário final. Dentro do Redshift, a técnica de agregação condicional é a ferramenta mais eficiente para construir essa tabela a partir de dados de log, superando alternativas como múltiplos JOINs.

Usando Funções de Ordem Superior (Higher-Order Functions - HOFs)

Richardson — Thu, 25 Sep 2025 00:41:27 +0000

Se você trabalha com PySpark e já precisou aplicar uma lógica dentro de uma coluna do tipo array, sua primeira reação provavelmente foi: "Vou criar uma UDF". É uma solução rápida e flexível, mas que esconde um grave problema de performance.

Neste post, vamos explorar a forma correta e muito mais eficiente de fazer isso usando Funções de Ordem Superior (Higher-Order Functions - HOFs).

O Custo Invisível das UDFs de Python

Uma User-Defined Function (UDF) em Python parece simples, mas por baixo dos panos, o Spark faz um trabalho caro:

Serialização: Para cada linha, os dados da coluna saem do ambiente otimizado do Spark (JVM).
Transferência: Os dados são enviados para um processo Python.
Execução: Sua função Python é executada.
Desserialização: O resultado volta da Python para a JVM.

Esse "ida e volta" para cada elemento de dados em um cluster é um gargalo gigante. Em datasets de milhões ou bilhões de linhas, isso pode fazer seu job demorar horas a mais ou até mesmo falhar.

A Solução: Funções de Ordem Superior (HOFs)

HOFs são funções nativas do Spark SQL que recebem outras funções (geralmente lambdas) como argumento para processar dados complexos, como arrays e mapas.

A grande vantagem é que toda a operação acontece dentro da JVM, sem o custo de serialização/desserialização. A lógica que você define na função lambda é executada pelo próprio motor do Spark, aproveitando toda a sua otimização.

Use HOFs quando precisar:

Transformar cada elemento de um array (transform).
Filtrar elementos de um array com base em uma condição (filter).
Verificar se um elemento que satisfaz uma condição existe no array (exists).
Agregar elementos de um array (aggregate).

Mão na Massa: Exemplos Práticos

Vamos criar um DataFrame simples para nossos exemplos. Nossa coluna scores é um array de inteiros.

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, col

# Criando a Spark Session
spark = SparkSession.builder.appName("HOF_Examples").getOrCreate()

# DataFrame de exemplo
data = [
    (1, "aluno_a", [80, 92, 75, 88]),
    (2, "aluno_b", [60, 70, 65, 58]),
    (3, "aluno_c", [95, 98, 100, 92]),
]
columns = ["id", "aluno", "scores"]
df_scores = spark.createDataFrame(data, columns)

1. `transform`: Aplicando uma transformação a cada elemento

df_bonus = df_scores.withColumn(
    "scores_com_bonus",
    expr("transform(scores, x -> x + 10)")
)
df_bonus.show(truncate=False)

2. `filter`: Filtrando elementos de um array

df_aprovados = df_scores.withColumn(
    "notas_altas",
    expr("filter(scores, nota -> nota >= 90)")
)
df_aprovados.show(truncate=False)

3. `exists`: Verificando a existência de um elemento

df_nota_max = df_scores.withColumn(
    "tirou_100",
    expr("exists(scores, nota -> nota = 100)")
)
df_nota_max.show()

Guia de Referência Rápida: Funções de Ordem Superior

Aqui está uma lista das HOFs mais comuns para você consultar.

Para Arrays

transform(array, function)
- Descrição: Aplica uma função a cada elemento do array e retorna um novo array com os resultados.
- Exemplo de Uso: Converter todos os nomes de um array para maiúsculas. transform(nomes, nome -> upper(nome))
filter(array, function)
- Descrição: Retorna um novo array contendo apenas os elementos que satisfazem uma condição booleana.
- Exemplo de Uso: Manter apenas os números pares de um array. filter(numeros, n -> n % 2 == 0)
exists(array, function)
- Descrição: Retorna true se pelo menos um elemento do array satisfaz a condição.
- Exemplo de Uso: Verificar se há algum produto com status "URGENTE" em um array de status. exists(status, s -> s = 'URGENTE')
forall(array, function)
- Descrição: Retorna true se todos os elementos do array satisfazem a condição.
- Exemplo de Uso: Checar se todas as tarefas de um projeto estão com status "CONCLUÍDO". forall(tarefas, t -> t.status = 'CONCLUÍDO')
aggregate(array, start, merge [, finish])
- Descrição: Reduz os elementos de um array a um único valor, começando com um valor inicial e aplicando uma função de merge.
- Exemplo de Uso: Somar todos os valores de um array de números. aggregate(valores, 0, (acumulador, valor) -> acumulador + valor)
zip_with(array1, array2, function)
- Descrição: Une dois arrays, elemento por elemento, aplicando uma função que combina os pares.
- Exemplo de Uso: Calcular o total de cada item multiplicando um array de quantidades por um de precos. zip_with(quantidades, precos, (q, p) -> q * p)

Para Mapas

transform_keys(map, function)
- Descrição: Aplica uma função a cada chave do mapa e retorna um novo mapa.
- Exemplo de Uso: Padronizar todas as chaves de um mapa para minúsculas. transform_keys(mapa, (k, v) -> lower(k))
transform_values(map, function)
- Descrição: Aplica uma função a cada valor do mapa e retorna um novo mapa.
- Exemplo de Uso: Aplicar um desconto de 10% em todos os preços de um mapa produto -> preco. transform_values(mapa, (k, v) -> v * 0.9)
map_filter(map, function)
- Descrição: Retorna um novo mapa contendo apenas as entradas que satisfazem uma condição.
- Exemplo de Uso: Filtrar um mapa de produto -> estoque para manter apenas produtos com estoque maior que zero. map_filter(mapa, (k, v) -> v > 0)

Experimente Você Mesmo!

Uma ótima maneira de testar tudo o que vimos aqui, sem a dor de cabeça de configurar um ambiente Spark local, é através da Databricks Free Edition, que oferece um cluster gratuito para estudo e desenvolvimento.

Para facilitar, deixei um notebook público com todo o código deste post pronto para ser visualizado e importado para a sua conta:

➡️ Notebook com os Exemplos: Visualizar no Databricks

Conclusão

Da próxima vez que você precisar manipular elementos dentro de um array (ou mapa) no PySpark, respire fundo e lembre-se das Funções de Ordem Superior.

Com os exemplos práticos e o guia de referência acima, você tem tudo o que precisa para começar. Pergunte-se: "Consigo resolver isso com transform, filter, exists ou outra HOF?". A resposta quase sempre será "sim", e seu pipeline de dados vai te agradecer com uma performance muito superior.

Paralelismo em Python para Engenharia de Dados: O Segredo das Tarefas I/O-Bound

Richardson — Sat, 06 Sep 2025 19:55:25 +0000

Você já escreveu um script para buscar dados de centenas de APIs ou ler milhares de arquivos e ficou olhando para o progresso, linha por linha, enquanto a maior parte do tempo o seu processador parecia estar de férias?

Se a resposta é sim, você provavelmente estava lidando com uma tarefa I/O-Bound. Em engenharia de dados, entender esse conceito é a chave para transformar pipelines lentos em processos eficientes.

Conceitos Fundamentais: Concorrência não é Paralelismo

Antes de mergulharmos no código, vale a pena esclarecer uma coisa. Como diz Rob Pike, um dos criadores da linguagem Go:

Concorrência é lidar com muitas coisas ao mesmo tempo.

Paralelismo é fazer muitas coisas ao mesmo tempo.

Nosso caso de I/O-bound é um exemplo clássico de concorrência. Nosso programa gerencia centenas de requisições de rede "pendentes" de uma só vez. Mesmo que tenhamos apenas um punhado de cores de CPU, o sistema operacional consegue dar progresso a cada uma delas, aproveitando o tempo de espera. Não estamos necessariamente executando os downloads em paralelo (o que exigiria centenas de cores), mas estamos gerenciando a concorrência entre eles de forma eficiente.

O Problema na Prática: A Execução Sequencial

A abordagem mais intuitiva para processar uma lista de tarefas é um loop for. Vamos simular a busca de dados para 200 produtos, onde cada busca leva 1 segundo (simulando a espera da rede).

import time

def buscar_dados_produto(product_id: int) -> str:
    """Simula uma chamada de rede que leva 1 segundo."""
    print(f"Buscando dados para o produto {product_id}...")
    time.sleep(1)
    return f"Dados do produto {product_id}"

# --- Execução Sequencial ---
inicio_seq = time.time()

resultados_seq = []
for i in range(200):
    resultados_seq.append(buscar_dados_produto(i))

fim_seq = time.time()
print(f"\nTempo total (Sequencial): {fim_seq - inicio_seq:.2f} segundos")

O cálculo do tempo é simples:

Tempo Total (Sequencial) = Número de Tarefas × Tempo por Tarefa
200 tarefas × 1s/tarefa = 200 segundos

É um processo lento, pois não aproveitamos o tempo de espera.

A Solução Concorrente: Threads ao Resgate

Se a CPU está ociosa enquanto espera, por que não usá-la para iniciar outras requisições? É exatamente isso que a concorrência com threads nos permite fazer. A biblioteca concurrent.futures do Python torna isso simples.

import time
import os
import concurrent.futures

def buscar_dados_produto(product_id: int) -> str:
    """Simula uma chamada de rede que leva 1 segundo."""
    print(f"Buscando dados para o produto {product_id}...")
    time.sleep(1)
    return f"Dados do produto {product_id}"

# --- Execução Paralela com Threads ---

# Identificar o número de cores para definir os workers dinamicamente
num_cores = os.cpu_count() or 1
# Para I/O-Bound, usamos um multiplicador. 4x o número de cores é um bom começo.
MAX_WORKERS = num_cores * 4
print(f"Configurando pool com {MAX_WORKERS} threads...")

inicio_par = time.time()

with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    # O executor.map agenda as tarefas e as executa nas threads.
    resultados_par = list(executor.map(buscar_dados_produto, range(200)))

fim_par = time.time()
print(f"\nTempo total (Threads): {fim_par - inicio_par:.2f} segundos")

O tempo de execução cai drasticamente. Para uma máquina com 4 cores (16 workers), o tempo teórico seria:

Tempo Total (Paralelo) ≈ (Número de Tarefas / Número de Workers) × Tempo por Tarefa
(200 tarefas / 16 workers) * 1s/tarefa ≈ 12.5 segundos

Gráfico Comparativo: Visualizando o Ganho

Este gráfico ilustra perfeitamente os conceitos que discutimos:

Ineficiência Sequencial: A linha "Tempo Sequencial" cresce de forma perfeitamente linear e íngreme. Dobrar o número de tarefas dobra o tempo de execução, como esperado. Em uma escala logarítmica, isso se manifesta como uma linha reta e diagonal.
Impacto Massivo do Paralelismo: Todas as linhas de execução paralela estão ordens de magnitude abaixo da linha sequencial, mostrando o ganho imediato e drástico de performance ao simplesmente não esperar em fila.
Benefícios de Mais Workers: A linha de 32 workers está consistentemente abaixo da de 16, que por sua vez está abaixo da de 8. Isso confirma que, para esta tarefa I/O-Bound, adicionar mais "trabalhadores" (threads) para fazer requisições concorrentes acelera ainda mais o processo.
Retornos Decrescentes: Note que a distância entre as linhas paralelas diminui. O salto de performance de "Sequencial" para "8 Workers" é gigantesco. O salto de "8" para "16" é ótimo, e o de "16" para "32" é bom, mas menor. Isso sugere que, em algum ponto, adicionar mais workers não trará um benefício tão grande, pois o sistema começará a ser limitado por outros fatores (largura de banda da rede, limites da API, etc.).

Por Que Threads e Não Processos? O GIL

O GIL: O Vilão que se Torna Herói no I/O

O Global Interpreter Lock (GIL) do Python é uma trava que permite que apenas uma thread execute bytecode Python por vez. Para tarefas que usam intensivamente a CPU, isso é um gargalo, pois impede o paralelismo real em múltiplos cores.

No entanto, para tarefas I/O-bound, o GIL se comporta de maneira diferente. O segredo é que toda função da biblioteca padrão do Python que faz uma chamada de sistema (syscall) libera o GIL.

Operações de rede e de disco são syscalls. Isso significa que quando a Thread A faz uma chamada para ler um arquivo do S3, ela libera a trava do GIL. Isso permite que a Thread B assuma o controle e inicie sua própria chamada de rede.

O resultado, como diz o autor David Beazley, é que "threads em Python são ótimas em não fazer nada" — e isso é exatamente o que queremos. Elas são a ferramenta perfeita para gerenciar a "espera" de forma concorrente. Usar processos (ProcessPoolExecutor) teria um custo de memória e inicialização muito maior, que é desnecessário para tarefas que não competem por tempo de CPU.

O Que Ferramentas Como o `ThreadPoolExecutor` Fazem por Nós?

Escrever código concorrente do zero é complexo. Ferramentas como concurrent.futures escondem essa complexidade de nós.

Gerenciamento do Ciclo de Vida: Iniciar threads tem um custo. Reutilizá-las em um "pool" é muito mais eficiente do que criar e destruir uma thread para cada pequena tarefa. O ThreadPoolExecutor faz exatamente esse gerenciamento de pool para nós.
Coordenação da Comunicação: Como obter o resultado de uma tarefa que rodou em outra thread? O método .map() do executor abstrai tudo isso, coletando os resultados e até mesmo tratando erros de forma transparente, sem que precisemos implementar filas ou outros mecanismos de comunicação manualmente.

Próximos Passos: Dask e Spark

Para tarefas de engenharia de dados mais complexas ou com volumes que não cabem na memória, as ferramentas da biblioteca padrão atingem seu limite. É aí que entram frameworks mais robustos.

Dask: Oferece uma abstração de alto nível sobre o paralelismo em Python, com DataFrames e Bags que podem operar em dados maiores que a memória e escalar para múltiplos nós.
Spark: É o padrão da indústria para processamento de Big Data. Com seu motor otimizado (Catalyst) e arquitetura distribuída, ele lida com transformações complexas (joins, agregações) em terabytes de dados de forma eficiente.

Exploraremos como essas ferramentas resolvem o mesmo problema (e muitos outros) em posts futuros.

Conclusão

Entender a natureza da sua carga de trabalho é o primeiro passo para a otimização. Muitas tarefas em engenharia de dados não são limitadas pela velocidade de processamento, mas pelo tempo de espera.

Da próxima vez que seu script de ingestão de dados parecer lento, pergunte-se: meu código está realmente trabalhando ou está apenas esperando? Se a resposta for "esperando", concurrent.futures.ThreadPoolExecutor é uma ferramenta simples e poderosa da biblioteca padrão para transformar essa espera em eficiência.

Referências

Fluent Python (Luciano Ramalho)

Automatize Suas Tarefas Diárias de Programação com Gemini CLI

Richardson — Tue, 26 Aug 2025 17:00:57 +0000

O Gemini CLI, com sua recente adição de comandos customizados e a capacidade de interagir com o shell, se tornou uma ferramenta indispensável no meu dia a dia. Neste post, vou mostrar como você pode usar o Gemini CLI, tanto com seus comandos internos quanto com scripts Bash, para automatizar algumas das tarefas mais comuns e liberar seu tempo para desafios mais interessantes.

Comando personalizado para analisar e explicar o código

Frequentemente encontra arquivos Python com funções complexas ou não documentadas. É necessário investir um tempo lendo manualmente o código para entender sua finalidade, parâmetros e valores de retorno antes de poder usá-lo ou modificá-lo com segurança.

Podemos criar um comando Gemini CLI reutilizável que possa analisar qualquer arquivo Python. Este comando deve gerar uma explicação clara e bem estruturada de cada função dentro daquele arquivo e fornecer uma maneira fácil de salvar essa análise para referência futura.

1.Criaremos um comando global com namespace chamado /py:explain dentro do diretório .gemini/commands. Isso permitirá que você o execute em qualquer diretório do projeto.

mkdir -p ~/.gemini/commands/py
touch ~/.gemini/commands/py/explain.toml

2.Agora, abra o arquivo explain.toml e adicione o seguinte prompt. Este prompt instrui o Gemini a atuar como um redator técnico especialista e fornece um formato estruturado para a saída.

# ~/.gemini/commands/py/explain.toml

description = "Analyzes a Python file and generates a detailed explanation of each function."
prompt = """
You are an expert Python programmer and technical writer, skilled at creating clear and concise documentation.

Please analyze the following Python code, which I am providing from the file `{{args}}`:

!{cat {{args}}}
For each function in this file, generate a detailed explanation in Markdown format. Follow this structure precisely for every function:

function_name()
Purpose: A single, clear sentence explaining what the function does.

Parameters:

param_name (type): Description of the parameter.

Returns:

(type): Description of what the function returns.

Example Usage:

Python

# A simple, self-contained code snippet showing how to use the function.
Notes: Mention any important details, potential edge cases, or dependencies.

"""

Como funciona:

description: Fornece o texto útil que você vê no menu de ajuda do Gemini CLI.

prompt: Este é o conjunto de instruções detalhadas para o modelo.
!{cat {{args}}}: Esta é a parte principal. Ele executa o comando shell cat no caminho do arquivo que você fornecer ({{args}}), injetando todo o conteúdo do arquivo diretamente no prompt para análise.

Agora você pode usar seu novo comando de dentro do Gemini CLI. Para salvar a saída diretamente em um arquivo, você pode executar gemini de forma não interativa no seu terminal e usar o redirecionamento padrão do shell.

Digamos que você queira entender um arquivo localizado em src/data_processing.py.

A interação com gemini cli pode ser feita através também através de scripts (bash). Aqui tem um exemplo para criar commits: https://github.com/richardson-souza/gemini-cli-automations/blob/main/.gemini/commands/git/git-atomic-commit.md

Referências

Testando com Monkey Patching

Richardson — Thu, 07 Aug 2025 20:31:35 +0000

O Cenário

Todo desenvolvedor já passou por isso: você precisa alterar ou dar manutenção em um trecho de código que não foi escrito pensando em testes. Frequentemente, esse código mistura lógica de negócio com configurações globais ou dependências implícitas, tornando a criação de testes unitários um desafio.

Um exemplo clássico, especialmente em pipelines de dados, é uma função que utiliza uma sessão Spark (spark) que existe como uma variável global no ambiente de produção, mas que não está definida no escopo de um teste local.

A solução ideal seria refatorar o código para usar Injeção de Dependência, mas nem sempre temos tempo ou permissão para fazer grandes alterações na base de código. Então, como criamos uma rede de segurança para garantir que nossas alterações funcionem? A resposta tática é Monkey Patching.

Definições Rápidas

Antes de prosseguir, vamos alinhar dois conceitos-chave:

Injeção de Dependência (Dependency Injection - DI): Um padrão de projeto onde as dependências de um componente (objetos, configurações, conexões) são fornecidas a ele externamente, em vez de serem criadas internamente. Na prática, significa "passar o que a função precisa como parâmetro".
Monkey Patching: Uma técnica que permite modificar ou substituir dinamicamente o comportamento de módulos, classes ou funções em tempo de execução. Em testes, usamos isso para substituir dependências reais (como bancos de dados ou APIs) por objetos falsos ("mocks").

O Código-Alvo

Imagine a seguinte função em um arquivo data_processor.py. Ela recebe um RDD, mas depende de um objeto spark que não está em sua assinatura.

# my_project/data_processor.py

from pyspark.sql import RDD, DataFrame
from pyspark.sql.functions import col, explode, split
import re

def process_raw_logs(log_rdd: RDD) -> DataFrame:
    """
    Transforma um RDD de logs brutos, quebrando JSONs concatenados
    e retornando um DataFrame estruturado.
    """
    # Mapeia o RDD para adicionar delimitadores
    mapped_rdd = log_rdd.map(
        lambda line: {"content": re.sub(r'\}\{', "}##!!##{", line)}
    )

    # CRASH! A variável 'spark' não existe no escopo do teste.
    df = spark.createDataFrame(mapped_rdd)

    df_processed = df.withColumn(
        "content", explode(split(col("content"), "##!!##"))
    )

    return df_processed

O Desafio do Teste

Um teste direto para essa função falharia, pois o spark não está definido.

# tests/test_data_processor.py

def test_process_raw_logs_fails(spark): # 'spark' aqui é uma fixture do pytest
    from my_project.data_processor import process_raw_logs

    # ... código para criar um RDD de teste ...

    # A linha abaixo irá falhar com: NameError: name 'spark' is not defined
    result = process_raw_logs(test_rdd)

A Solução: Monkey Patching em Ação

Para contornar isso, vamos usar o monkeypatch do pytest para injetar a sessão spark (fornecida pela nossa fixture de teste) diretamente no módulo data_processor antes de chamar a função.

# tests/test_data_processor.py

from pyspark.sql import SparkSession

def test_process_raw_logs_with_monkeypatch(spark: SparkSession):
    # 1. Importamos o módulo que queremos testar, não a função diretamente.
    #    Isso nos dá um objeto para "remendar".
    from my_project import data_processor

    # 2. AQUI ESTÁ O TRUQUE: Monkey Patching.
    #    Criamos um atributo chamado 'spark' dentro do módulo 'data_processor'
    #    e atribuímos a ele a nossa fixture 'spark' do teste.
    data_processor.spark = spark

    # 3. Agora, preparamos nosso cenário de teste.
    log_data = ['{"id": 1}{"id": 2}', '{"id": 3}']
    test_rdd = spark.sparkContext.parallelize(log_data)

    # 4. Executamos a função.
    #    Quando a função `process_raw_logs` procurar por 'spark', ela o encontrará
    #    no escopo do seu próprio módulo, pois nós o colocamos lá.
    result_df = data_processor.process_raw_logs(test_rdd)

    # 5. Verificamos o resultado.
    assert result_df.count() == 3
    assert "content" in result_df.columns

    # O pytest garante que essa modificação no módulo seja desfeita após o teste,
    # evitando contaminação entre testes.

Por Que Isso Funciona?

Em Python, módulos são objetos. Quando você faz import my_project.data_processor, você está carregando o código daquele arquivo em um objeto de módulo na memória. O que a linha data_processor.spark = spark faz é simplesmente adicionar um novo atributo a esse objeto. A função process_raw_logs, ao ser executada, resolve o nome spark procurando primeiro em seu escopo local e depois no escopo do módulo onde foi definida, encontrando a nossa versão injetada.

Análise da Abordagem

Esta técnica é poderosa, mas deve ser usada com cautela.

Prós:
- Permite testar código que de outra forma seria "intestável".
- Serve como uma "rede de segurança" essencial para permitir futuras refatorações. Você pode escrever um teste como este para garantir que não quebrou nada ao fazer uma alteração.
Contras:
- É um "code smell". O teste se torna mais complexo e menos legível.
- A dependência da função continua oculta. Um desenvolvedor precisa ler o teste para entender que a função process_raw_logs depende de spark.
- O teste fica fortemente acoplado à estrutura do arquivo, não ao contrato da função.

Conclusão

Enfrentar código legado ou não projetado para testes é uma realidade. Embora a Injeção de Dependência seja o objetivo estratégico para um código limpo e manutenível, o Monkey Patching é uma ferramenta tática indispensável. Ele nos permite criar uma rede de segurança, garantindo a qualidade e a estabilidade do software enquanto pavimentamos o caminho para futuras melhorias. Use-o como um meio para um fim, não como o padrão final.

Referências

Pytest monkeypatch fixture: pytest.org documentation
unittest.mock (Standard Library): docs.python.org
Dependency Injection - Martin Fowler: martinfowler.com
Working Effectively with Legacy Code (Livro de Michael Feathers): Um livro fundamental sobre as estratégias discutidas aqui, como "Characterization Tests" e "Seams".

Pytest: Como Testar Módulos Python com Configuração no Nível Superior

Richardson — Wed, 06 Aug 2025 23:31:13 +0000

Introdução

Em engenharia de dados, é comum criar scripts procedurais em Python para orquestrar pipelines. Um padrão frequente nesses scripts é a definição de configurações globais — como nomes de buckets — no nível superior do módulo, baseadas em variáveis de ambiente que mudam entre dev e prd.

Embora seja uma abordagem direta, ela introduz um desafio significativo para a criação de testes unitários com pytest. Frequentemente, os testes falham durante a fase de importação, antes da execução de qualquer lógica, devido a configurações de ambiente ausentes.

Este artigo apresenta uma análise técnica da causa raiz desse problema, explica o papel fundamental do conftest.py e oferece duas soluções práticas para garantir que seus módulos sejam testáveis.

O Padrão de Código e o Desafio do Teste

Considere o script data_processor.py. Sua função é ler dados de um bucket "raw" e escrevê-los em um bucket "processed". Os nomes dos buckets são determinados pela variável de ambiente DEPLOY_ENV.

data_processor.py

import os
# Suponha uma biblioteca interna para operações S3
import s3_utils 

# 1. Configuração lida no nível do módulo
ENV_NAME = os.getenv("DEPLOY_ENV")  # Ex: 'dev' ou 'prd'

# 2. Validação que ocorre durante a importação
if not ENV_NAME:
    raise ValueError("A variável de ambiente DEPLOY_ENV não foi definida.")

# 3. Variáveis globais construídas a partir da configuração
RAW_DATA_BUCKET = f"company-data-{ENV_NAME}-raw"
PROCESSED_DATA_BUCKET = f"company-data-{ENV_NAME}-processed"

def process_source_file(source_id: str) -> dict:
    """
    Lê um arquivo da zona raw, processa e salva na zona processed.
    """
    source_path = f"s3://{RAW_DATA_BUCKET}/sources/{source_id}.csv"
    destination_path = f"s3://{PROCESSED_DATA_BUCKET}/reports/{source_id}.parquet"

    data = s3_utils.read_csv(source_path)
    # ...lógica de processamento...
    s3_utils.write_parquet(data, destination_path)

    return {"source": source_path, "destination": destination_path}

O desafio é claro: a linha raise ValueError será executada assim que o módulo for importado se DEPLOY_ENV não estiver definida, impedindo qualquer teste.

A Estratégia de Teste Inicial: Usando `conftest.py`

Para testar nosso script, precisamos controlar suas dependências externas. Neste caso, a dependência é a variável de ambiente DEPLOY_ENV. A ferramenta padrão e mais poderosa do pytest para gerenciar configurações e dependências compartilhadas é o arquivo conftest.py.

O que é o conftest.py?
É um arquivo especial que o pytest procura e carrega automaticamente. Ele permite definir fixtures, que são funções de setup e teardown reutilizáveis. Tudo que é definido em um conftest.py fica disponível para todos os testes no mesmo diretório e em subdiretórios.

Nossa estratégia inicial seria criar um conftest.py para carregar um ambiente de teste.

tests/conftest.py

import pytest
from dotenv import load_dotenv

@pytest.fixture(scope="session", autouse=True)
def load_test_environment():
    """
    Carrega variáveis de um arquivo .env para a sessão de testes.
    """
    load_dotenv(dotenv_path="tests/.env.test")

Vamos detalhar essa fixture:

@pytest.fixture: Transforma a função em uma fixture do pytest.
scope="session": Define que a fixture será executada apenas uma vez por sessão de teste, e não antes de cada teste. É ideal para configurações que não mudam.
autouse=True: Este é o parâmetro chave. Ele instrui o pytest a executar esta fixture automaticamente para todos os testes, sem que precisemos solicitá-la explicitamente. É perfeito para um setup de ambiente global.

Com essa configuração, parece que nosso problema está resolvido. No entanto, o seguinte teste ainda falhará:

tests/test_processor_fail.py

# Este import irá disparar a validação em data_processor.py
from data_processor import process_source_file

def test_file_processing(mocker):
    # O erro ocorre antes que o corpo do teste seja executado.
    ...

Análise da Causa Raiz: Ordem de Execução no Pytest

Apesar de nossa configuração correta no conftest.py, a falha ocorre devido à interação entre o mecanismo de importação do Python e o ciclo de vida do pytest.

Setup da Sessão Pytest: pytest inicia e executa nossa fixture load_test_environment devido ao scope="session" e autouse=True. O ambiente de teste, com DEPLOY_ENV=dev, é carregado.
Coleta de Testes: Em seguida, pytest inicia a fase de coleta. Ele encontra tests/test_processor_fail.py e o interpretador Python executa a instrução from data_processor import process_source_file.
Falha na Importação: É neste momento que o código no nível superior do data_processor.py é executado. Por razões de "timing" e isolamento de processos na fase de coleta, o ambiente recém-configurado pela fixture pode não estar visível para este processo de importação imediato, resultando no ValueError.
Execução do Teste: A fase de execução do teste nunca é alcançada.

A solução é garantir que a importação do módulo problemático ocorra somente após o ambiente de teste do pytest estar completamente estabelecido e visível.

Solução 1: Importação Local

A abordagem mais direta é mover a instrução import do topo do arquivo para dentro da função de teste.

tests/test_processor_solution1.py

def test_process_source_file_with_local_import(mocker):
    """
    Testa o processamento de arquivos adiando a importação do módulo.
    """
    # 1. A importação ocorre aqui, dentro do escopo de execução do teste.
    from data_processor import process_source_file

    # 2. Agora podemos mockar as dependências do módulo recém-importado.
    mock_s3_utils = mocker.patch("data_processor.s3_utils")

    # 3. Executamos a função e validamos o resultado.
    result = process_source_file("user_123")

    expected_dest = "s3://company-data-dev-processed/reports/user_123.parquet"
    mock_s3_utils.write_parquet.assert_called_once_with(mocker.ANY, expected_dest)
    assert result["destination"] == expected_dest

Vantagens: Simples, explícito e resolve o problema de forma eficaz.
Desvantagens: Pode levar à repetição do import se múltiplos testes no mesmo arquivo precisarem da mesma função.

Solução 2: Fixture para Injeção do Módulo

Uma alternativa mais escalável é encapsular a importação local dentro de uma fixture. O teste então declara sua dependência nesta fixture, que fornece o módulo importado como um objeto. Para maior clareza, esta fixture pode ser definida no próprio arquivo de teste.

tests/test_processor_solution2.py

import pytest

@pytest.fixture(scope="module")
def data_processor_module():
    """
    Fixture que importa e retorna o módulo data_processor.
    A importação ocorre apenas quando a fixture é utilizada.
    """
    import data_processor
    return data_processor

def test_process_source_file_with_fixture(mocker, data_processor_module):
    """
    Testa o processamento de arquivos usando um módulo injetado via fixture.
    """
    # 1. A fixture `data_processor_module` é executada, importando o módulo.
    mock_s3_utils = mocker.patch("data_processor.s3_utils")

    # 2. Chamamos a função através do objeto do módulo injetado.
    result = data_processor_module.process_source_file("user_123")

    expected_dest = "s3://company-data-dev-processed/reports/user_123.parquet"
    mock_s3_utils.write_parquet.assert_called_once_with(mocker.ANY, expected_dest)
    assert result["destination"] == expected_dest

Vantagens: Promove a reutilização de código (DRY), mantém os testes limpos e centraliza a lógica de importação tardia.
Desvantagens: Adiciona um nível de indireção que pode ser menos óbvio para desenvolvedores não familiarizados com o padrão.

Conclusão e Recomendações

O acoplamento entre a lógica de um módulo e sua configuração no nível superior é um desafio comum para a testabilidade. Compreender o ciclo de vida do pytest e o papel do conftest.py é fundamental para diagnosticar e resolver os problemas de importação resultantes.

A importação local (Solução 1) é a abordagem mais direta e recomendada para casos simples.
O uso de uma fixture para injeção (Solução 2) é preferível em cenários onde múltiplos testes precisam acessar diferentes funções de um mesmo módulo, oferecendo uma solução mais limpa e organizada.

Ambas as técnicas são ferramentas valiosas para aumentar a cobertura de testes e a robustez de aplicações Python que não foram inicialmente projetadas com a testabilidade em mente.

What Is Change Data Capture (CDC) and How It Works on Google Cloud

Richardson — Sun, 13 Jul 2025 18:52:55 +0000

Keeping analytics pipelines real‑time and resource‑efficient is table stakes in 2025.

That’s where Change Data Capture (CDC) shines—streaming only the rows that changed instead of bulk‑copying entire tables.

In this post, you’ll learn:

What CDC is (and why it matters).
Three core implementation patterns—query‑based, trigger‑based, log‑based.
How to pick the right Google Cloud tool—Datastream or a DIY Debezium + Dataflow combo.
A quick‑reference exam tip for anyone chasing the Google Cloud Professional Data Engineer cert.

🔁 What Exactly Is CDC?

Change Data Capture (CDC) = detecting inserts, updates, and deletes in a source database and pushing just those deltas downstream (e.g., into BigQuery).

Result:

No heavy full‑table copies.
Near–real‑time dashboards and ML features.
Lower source‑DB load.

🔍 CDC Patterns Every Engineer Should Know

1. Query‑Based CDC (Timestamp / Version Column)

SELECT *
FROM customers
WHERE last_updated_timestamp > '2025-07-12 21:20:00';

✅ Pros	❌ Cons
Dead‑simple scripting	Misses deletes
No extra tooling	Adds query load
	Requires schema change (extra column)

2. Trigger‑Based CDC

Flow: Triggers (AFTER INSERT/UPDATE/DELETE) copy changes into a *_history table.

✅ Pros	❌ Cons
Captures all ops, incl. deletes	High write‑time overhead
Built‑in audit trail	Harder to maintain at scale

3. Log‑Based CDC (Modern Standard)

Reads the DB’s transaction log (MySQL binlog, Postgres WAL, etc.).

✅ Pros	❌ Cons
Low latency (near real time)	Needs specialized tool
Minimal DB impact	Setup can be tricky
Captures deletes & schema changes

🚀 Implementing CDC on Google Cloud

Option A — Datastream (Managed, Serverless)

Log‑based CDC for MySQL, Postgres, Oracle, AlloyDB.
Streams raw events into BigQuery staging tables.
Auto‑executes MERGE so target tables stay current.
Handles schema drift for you.

Perfect when you want “set it and forget it” replication.

Option B — Debezium + Pub/Sub + Dataflow (DIY Flex)

Debezium connectors tail the transaction log.
Changes land in Pub/Sub.
Dataflow applies custom transforms → BigQuery.

Use this path when you need complex, in‑flight transformations or to support a niche source DB Debezium already speaks.

🎓 Exam Tip

For the Google Cloud PDE exam, default to Datastream for relational‑to‑BigQuery CDC.
Reach for Dataflow + Debezium only if the scenario explicitly calls for heavy transformations or bespoke routing.

🧠 Pattern Cheat‑Sheet

Pattern	Captures Deletes?	Source DB Load	Complexity	GCP Tool of Choice
Query‑Based	❌	🔺 Medium	🟢 Low	N/A (custom script)
Trigger‑Based	✅	🔺 High	🔺 Medium	N/A (DB triggers)
Log‑Based	✅	🟢 Low	🔺 High	Datastream, Debezium

Ready to Build?

CDC turns stale ETL batches into streaming insights with surprisingly little effort—especially with Datastream doing the heavy lifting.

Have questions, war stories, or tips? Drop them below—let’s level‑up together 💬

Thanks for reading! If you found this helpful, consider following me for more posts on data engineering, GCP, and real‑world pipeline design.

Um Projeto Prático para Estudar RAG: Análise Qualitativa de Código com LLMs Locais

Richardson — Sun, 15 Jun 2025 03:43:44 +0000

Introdução

Procurar informações em um repositório de código gigante pode ser frustrante. A resposta muitas vezes está ali, mas perdida em meio a centenas de arquivos. Ferramentas como grep resolvem buscas exatas, mas e se precisássemos de uma interpretação, uma "conversa" com o código?

Para explorar essa questão, desenvolvi este projeto. O objetivo não é criar uma ferramenta de produção, mas sim servir como um laboratório prático para apoiar os estudos sobre LLMs e a arquitetura RAG. É uma oportunidade de observar, com um exemplo concreto, como esses componentes funcionam juntos.

O foco do projeto foi criar um sistema capaz de responder a perguntas como:

"Quem é o responsável pela DAG X?"
"Qual a finalidade da função Y?"
"Mostre-me as queries que utilizam a tabela XYZ."

O resultado foi uma ferramenta de linha de comando funcional e, mais importante, um grande aprendizado que compartilho a seguir.

Parte 1: A Ideia - RAG e o Papel Qualitativo do LLM

A base do projeto é a arquitetura RAG (Retrieval-Augmented Generation), que une uma busca eficiente com a capacidade de interpretação de um Modelo de Linguagem Grande (LLM).

O processo funciona em três etapas:

A Base de Conhecimento (Vector Store): Primeiro, o conteúdo do repositório é indexado. Um script lê, divide os arquivos em trechos (chunks) e usa um modelo de embedding para converter cada trecho em vetores numéricos, que são armazenados localmente.
O Detetive (Retriever): Quando uma pergunta é feita, o sistema busca na base de vetores os trechos de código mais relevantes. Esta é a etapa de recuperação de dados, puramente quantitativa: encontrar a informação exata.
O Intérprete (LLM): Aqui está o ponto crucial e o grande diferencial deste projeto. Os trechos de código encontrados são entregues ao LLM junto com a pergunta original. É neste momento que a mágica acontece.

LLMs para Respostas Qualitativas

Enquanto a busca (retriever) encontra o "o quê" (o trecho de código, a linha exata), o LLM é excepcional em fornecer o "porquê" e o "como". Sua força não está em encontrar dados, mas em sintetizar, resumir, explicar e inferir informações a partir do contexto fornecido. Ele transforma dados brutos em respostas qualitativas, que se assemelham à interpretação que um ser humano faria.

O melhor de tudo? Todo o processo acontece localmente na sua máquina, tornando-o um ambiente perfeito e sem custos para estudos e experimentos.

Parte 2: Mãos à Obra - Preparando o Ambiente

Vamos reunir as ferramentas necessárias para o nosso projeto.

Ambiente Python

Primeiro, certifique-se de ter o Python 3.8 (ou superior) instalado. Em seguida, crie uma pasta para o projeto e um ambiente virtual para manter as dependências organizadas.

mkdir code-qa-bot # Ou o nome que preferir
cd code-qa-bot
python3 -m venv .venv
source .venv/bin/activate

Ollama: IA na sua Máquina

O Ollama é a forma mais simples de rodar modelos de linguagem localmente. Baixe e instale a versão para o seu sistema operacional. Depois, via terminal, baixe o modelo que usaremos para gerar as respostas. Usaremos o gemma:2b, um modelo do Google leve e competente.

ollama pull gemma:2b

Abaixo, uma tabela comparativa para ajudar na escolha de outros modelos, caso queira experimentar:

Model Name	Size	Context Length	Speed	Quality (Code/NL)	RAM (Quantized)	Best Use Case	License
Phi-3 Mini	3.8B	4K	⚡ Very Fast	🟢 Excellent NL	~4GB (Q4)	General Q&A, search, chat	MIT
Gemma 2B	2B	8K	⚡ Very Fast	🟡 Moderate code	~3.5GB (Q4)	Lightweight assistants	Google
Mistral 7B	7B	8K	⚠️ Moderate	🟢 Strong code+NL	~8–9GB (Q4)	General-purpose, coding	Apache 2.0
TinyLlama 1.1B	1.1B	2K	🚀 Extremely Fast	🟡 Limited NL/code	~2GB (Q4)	Embedded tools, CLI help	Apache 2.0
LLaMA 3 8B	8B	8K	⚠️ Moderate	🟢 Very strong NL/code	~9–10GB (Q4)	High-quality retrieval + logic	Meta
OpenChat 3.5 7B	7B	8K	⚠️ Moderate	🟢 Very accurate code	~8GB (Q4)	Coding-focused assistant	Apache 2.0

Bibliotecas Python

Crie um arquivo requirements.txt com as seguintes dependências. Você pode ver o arquivo original aqui.

# Orquestração principal da pipeline RAG
langchain
langchain_community

# Ferramenta para rodar LLMs locais
ollama

# Integrações para componentes específicos
langchain-huggingface  # Para o modelo de embedding
langchain-chroma       # Para o vector store ChromaDB
langchain-ollama       # Para conectar ao LLM local via Ollama

# Provedor do modelo de embedding
sentence-transformers

# Banco de dados vetorial local
chromadb

# Utilitários para processamento de arquivos
PyYAML              # Para arquivos .yaml
sql-metadata        # Para arquivos .sql
lark

pytest

Agora, instale todas as bibliotecas de uma vez:

pip install -r requirements.txt

Parte 3: O Código - Construindo o Motor de Busca

Nossa ferramenta será modular, dividida em vários arquivos Python para maior clareza.

config.py
Este arquivo centraliza as configurações. A principal alteração que você deve fazer aqui é apontar a variável REPO_PATH para o caminho do seu repositório local.
data_loader.py
Aqui está o coração do pré-processamento. Em vez de tratar todos os arquivos como texto genérico, este módulo os analisa para extrair informações estruturadas e valiosas. É aqui que definimos o page_content e os metadata de cada "documento".

⚠️ Atenção: Ponto Crucial de Customização!
As classes de processamento (YamlProcessor, SqlProcessor, etc.) foram desenhadas para uma estrutura de projeto específica. O seu repositório provavelmente terá uma organização diferente.

Pense neste código como um template. A estratégia fundamental é criar "chunks" de informação inteligentes. Você precisará adaptar a lógica dentro de cada classe process para que ela entenda e extraia as informações mais relevantes do seu contexto.

A importância de page_content e metadata

Definir bem esses dois parâmetros é o segredo para uma busca precisa. Pense neles como uma ficha de catalogação de uma biblioteca:
- page_content (O Conteúdo do Livro): É o texto que será efetivamente "lido" e vetorizado. Um page_content claro e rico em contexto gera uma representação vetorial muito mais fiel. Por exemplo, a frase "Este documento descreve a DAG com ID 'dag_exemplo'. O proprietário é 'ana.silva'." tem um significado semântico muito mais forte do que um bloco YAML bruto. É também esse conteúdo que o LLM usará para formular a resposta final.
- metadata (A Etiqueta na Lombada): São os dados que descrevem o conteúdo, como dag_id, table_name, author, etc. A função mais poderosa do metadata é permitir a filtragem inteligente. Quando usamos um SelfQueryRetriever, ele primeiro usa os metadados para filtrar os documentos relevantes e só então faz a busca por similaridade semântica. Isso torna a busca dramaticamente mais rápida e precisa, evitando que o sistema se confunda com informações de arquivos não relacionados.
vector_store.py

Este arquivo gerencia a criação e o carregamento do nosso banco de dados vetorial persistente (ChromaDB).
indexer.py

Um script independente que você executa uma única vez (ou sempre que houver mudanças significativas no código) para popular o banco de dados vetorial.
ask.py

Este é o script interativo para fazer perguntas ao seu código. Ele configura o Self-Querying Retriever, o cérebro da nossa operação. Graças aos metadados ricos que definimos, o retriever usa o LLM para analisar sua pergunta e criar um filtro preciso antes mesmo de realizar a busca vetorial.

Parte 4: O Fluxo de Trabalho na Prática

Com todos os arquivos no lugar, usar a ferramenta se resume a duas etapas.

1. Indexe seu Repositório

Este passo único lê todos os arquivos do seu projeto, os processa e constrói o banco de dados vetorial.

❯ python indexer.py
Starting the indexing process...
Found and processed 19052 document chunks.
Creating new vector store...
Vector store created and saved.
✅ Indexing complete.

2. Faça suas Perguntas

Agora, inicie a ferramenta de Q&A e comece a conversar com seu código.

Exemplo 1: Pergunta sobre responsabilidade

❯ python ask.py
✅ Q&A Tool is ready. Ask questions about your codebase.
Ask a question (or type 'exit'): Who is the owner of the dag "process_seller_data_dag"?

--- Answer ---
ana.silva@suaempresa.com

--- Sources ---
- ./data/dags/process_seller_data_dag/dag.yaml

Ask a question (or type 'exit'):

Exemplo 2: Pergunta sobre uso de tabelas

❯ python ask.py
✅ Q&A Tool is ready. Ask questions about your codebase.

Ask a question (or type 'exit'): Quais consultas utilizam a tabela 'analytics_prod.seller_reputation'?

--- Answer ---
Com base nos documentos fornecidos, a tabela `analytics_prod.seller_reputation` é utilizada principalmente na seguinte consulta SQL, encontrada no arquivo `classificacao_seller.sql`:

`SELECT ... FROM analytics_prod.seller_reputation WHERE ...`

Esta query parece ser central para calcular a classificação e o score dos vendedores.

Além disso, podemos inferir outros usos potenciais baseados na estrutura do projeto:

1.  **Relatórios de Performance:** A tabela é provavelmente uma fonte de dados para dashboards e relatórios que monitoram a performance dos vendedores.
2.  **Análises de Dados:** Consultas de agregação (`GROUP BY`) podem ser usadas para analisar tendências e padrões no comportamento dos vendedores.
3.  **Lógica de Negócio:** Os dados podem disparar processos automáticos, como o envio de notificações para vendedores com base em sua performance.

Para uma lista exaustiva, seria necessário analisar todas as dependências que consomem os outputs desta DAG.

--- Sources ---
- ./data/dags/process_seller_data_dag/assets/json/classificacao_seller.json
- ./data/dags/process_seller_data_dag/readme.md
- ./data/dags/process_seller_data_dag/sql/taxa_score/classificacao_seller.sql
- ./data/dags/process_seller_data_dag/sql/taxa_score/sellers.sql

Ask a question (or type 'exit'):


+---------------------------+
|   Pergunta do Usuário     |
| (Ex: "Descreva a DAG...") |
+---------------------------+
             |
             v
+--------------------------------------+
| 1. O Retriever busca os documentos   |
|    mais relevantes no Vector Store.  |
+--------------------------------------+
             |
             v
+--------------------------------------+
| 2. O conteúdo dos documentos         |
|    (page_content) é enviado para o   |
|    LLM como contexto.                |
+--------------------------------------+
             |
             v
+--------------------------------------+
| 3. O LLM lê o contexto e gera uma    |
|    resposta em texto (resumo).       |
+--------------------------------------+
             |
             v
+---------------------------+
|   Resposta para o Usuário |
+---------------------------+

Repositório no github

DEV Community: Richardson

Estacionariedade: Por que a Média Histórica é Perigosa para suas Projeções

Introdução

O Conceito: Raiz Unitária e Estacionariedade

O Diagnóstico: Teste Augmented Dickey-Fuller (ADF)

Exemplo Prático em Python

Resultados Obtidos

A Solução: Diferenciação no SARIMAX

Implementação no SARIMAX

Conclusão

Referências e Bibliografia

Regressão Linear para Inferência Causal: Indo Além da Predição

1. Diferença entre Predição e Inferência

2. Estudo de Caso: Qualidade de Cadastro no E-commerce

3. Outras Aplicações Técnicas

4. Limitações e Boas Práticas

Referências Bibliográficas

Guia arquitetônico de ponta para a construção de uma plataforma de dados

Etapa 1: A Conexão Fundamental - Modelo Lógico/Físico e a Arquitetura Medalhão

Etapa 2: A Estrutura Detalhada das Camadas no Google Cloud

Camada Bronze: O Data Lake Bruto e Imutável

Camada Prata: A Fonte da Verdade Normalizada e Confiável

Camada Ouro: Os Produtos de Dados Focados no Negócio

Etapa 3: Práticas Transversais Essenciais

Etapa 4: A Síntese Revisada - O Fluxo de Valor Orquestrado

Otimizando Redshift na Prática: Um Estudo de Caso com DISTKEY e SORTKEY

Parte 1: Conceitos Fundamentais

A. Chaves de Distribuição (DISTKEY)

B. Chaves de Ordenação (SORTKEY)

Parte 2: A Ferramenta de Diagnóstico: svv_table_info

Parte 3: O Ciclo de Otimização - Estudo de Caso

Etapa 1: Análise do Padrão de Consulta

Etapa 2: Diagnóstico da Tabela Original

Etapa 3: Definição da Estratégia de Otimização

Etapa 4: Implementação (Processo "Deep Copy")

Conclusão

Modelagem de Dados para Análise de Funis no Amazon Redshift

O Cenário: Funil de Tickets de Suporte

Decisão 1: A Arquitetura da Tabela - OBT vs. Star Schema

Decisão 2: A Técnica de Pivotagem no Redshift

Abordagem 1: Agregação Condicional (Recomendada)

Abordagem 2: Múltiplos Self-Joins (Não Recomendada)

Alternativa: Funções de Janela (Window Functions)

Dicas Finais de Otimização no Redshift

Conclusão

Usando Funções de Ordem Superior (Higher-Order Functions - HOFs)

O Custo Invisível das UDFs de Python

A Solução: Funções de Ordem Superior (HOFs)

Mão na Massa: Exemplos Práticos

1. transform: Aplicando uma transformação a cada elemento

2. filter: Filtrando elementos de um array

3. exists: Verificando a existência de um elemento

Guia de Referência Rápida: Funções de Ordem Superior

Para Arrays

Para Mapas

Experimente Você Mesmo!

Conclusão

Paralelismo em Python para Engenharia de Dados: O Segredo das Tarefas I/O-Bound

Conceitos Fundamentais: Concorrência não é Paralelismo

O Problema na Prática: A Execução Sequencial

A Solução Concorrente: Threads ao Resgate

Gráfico Comparativo: Visualizando o Ganho

Por Que Threads e Não Processos? O GIL

O GIL: O Vilão que se Torna Herói no I/O

O Que Ferramentas Como o ThreadPoolExecutor Fazem por Nós?

Próximos Passos: Dask e Spark

Conclusão

Referências

Automatize Suas Tarefas Diárias de Programação com Gemini CLI

Comando personalizado para analisar e explicar o código

Como funciona:

Referências

Testando com Monkey Patching

O Cenário

Definições Rápidas

O Código-Alvo

O Desafio do Teste

A Solução: Monkey Patching em Ação

Por Que Isso Funciona?

Análise da Abordagem

Parte 2: A Ferramenta de Diagnóstico: `svv_table_info`

Alternativa: Funções de Janela (`Window Functions`)

1. `transform`: Aplicando uma transformação a cada elemento

2. `filter`: Filtrando elementos de um array

3. `exists`: Verificando a existência de um elemento

O Que Ferramentas Como o `ThreadPoolExecutor` Fazem por Nós?

A Estratégia de Teste Inicial: Usando `conftest.py`