Pipeline ETL Avançado

Sistema ETL modular e extensível para processamento de dados em larga escala.

Características

• Suporte a múltiplas fontes de dados (CSV, JSON, Excel, Parquet, PostgreSQL, MySQL)
• Processamento paralelo e batch otimizado
• Validação de qualidade e schema
• Carga incremental e CDC
• Data lineage tracking
• Sistema de checkpoint e recuperação
• Exportação para múltiplos formatos
• Testes completos (80%+ cobertura)

Instalação

pip install -r requirements.txt

Uso Básico

Pipeline Simples

from etl.orchestrator import ETLOrchestrator

orchestrator = ETLOrchestrator("Meu Pipeline")
result = orchestrator.execute(
    source='data/input.csv',
    destination='output/processed.csv',
    source_type='csv',
    dest_type='csv'
)

print(f"Status: {result['status']}")
print(f"Linhas processadas: {result['rows_processed']}")

Pipeline com Transformações

from etl.orchestrator import ETLOrchestrator

def remover_duplicatas(df):
    return df.drop_duplicates()

def filtrar_ativos(df):
    return df[df['ativo'] == True]

orchestrator = ETLOrchestrator("Pipeline Transformado")
result = orchestrator.execute(
    source='data/usuarios.csv',
    destination='output/usuarios_limpos.parquet',
    source_type='csv',
    dest_type='parquet',
    transformations=[remover_duplicatas, filtrar_ativos]
)

Validação de Qualidade

from etl.quality import DataQualityValidator
import pandas as pd

df = pd.read_csv('data/clientes.csv')

validator = DataQualityValidator()
validator.add_completeness_check(['email', 'telefone'], threshold=0.95)
validator.add_uniqueness_check(['cpf'])
validator.add_pattern_check('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')

report = validator.validate(df)
print(f"Taxa de sucesso: {report['success_rate']:.1%}")

Validação de Schema

from etl.schema import TableSchema, ColumnSchema, DataType

schema = TableSchema('usuarios')
schema.add_column(ColumnSchema('id', DataType.INTEGER, nullable=False, unique=True))
schema.add_column(ColumnSchema('nome', DataType.STRING, nullable=False))
schema.add_column(ColumnSchema('idade', DataType.INTEGER, min_value=0, max_value=120))

result = schema.validate(df)
if not result['valid']:
    print(f"Erros: {result['errors']}")

Carga Incremental

from etl.incremental import IncrementalLoadManager
import pandas as pd

manager = IncrementalLoadManager('state/state.json')

df = pd.read_csv('data/vendas.csv')
incremental_df = manager.get_incremental_data(
    df=df,
    key_column='id',
    comparison_column='data_venda',
    state_key='vendas_pipeline'
)

print(f"Novos registros: {len(incremental_df)}")

Processamento Paralelo

from etl.parallel import ParallelProcessor
import pandas as pd

def processar_chunk(chunk):
    chunk['total'] = chunk['quantidade'] * chunk['preco']
    return chunk

processor = ParallelProcessor(n_workers=4)
df = pd.read_csv('data/pedidos.csv')

result_df = processor.process_chunks(df, processar_chunk)

Banco de Dados

from etl.database import DatabaseFactory

config = {
    'host': 'localhost',
    'port': 5432,
    'database': 'etl_db',
    'user': 'etl_user',
    'password': 'senha'
}

connector = DatabaseFactory.create('postgres', config)

with connector.transaction():
    df = connector.execute_query("SELECT * FROM vendas WHERE data > '2021-01-01'")
    connector.insert_dataframe(df_processado, 'vendas_processadas')

Checkpoint e Recuperação

from etl.checkpoint import CheckpointManager, PipelineRecovery

checkpoint_mgr = CheckpointManager('checkpoints/')

checkpoint_id = checkpoint_mgr.create_checkpoint(
    pipeline_name='vendas_pipeline',
    step='transformacao',
    data=df_transformado,
    metadata={'rows': len(df_transformado)}
)

recovery = PipelineRecovery(checkpoint_mgr)
if recovery.can_recover('vendas_pipeline'):
    checkpoint = recovery.recover('vendas_pipeline')
    df = checkpoint['data']

Data Lineage

from etl.lineage import DataLineageTracker

tracker = DataLineageTracker('lineage/lineage.json')

tracker.track_pipeline_execution(
    source='data/vendas.csv',
    transformations=['limpeza', 'agregacao', 'enriquecimento'],
    target='output/vendas_final.parquet',
    metadata={'author': 'pipeline_vendas', 'version': '2.0'}
)

tracker.save()
print(tracker.visualize_graph())

Exportação Multi-formato

from etl.export import DataExporter
import pandas as pd

df = pd.read_csv('data/relatorio.csv')

exporter = DataExporter()
results = exporter.export_multiple_formats(
    df=df,
    base_path='output/relatorio',
    formats=['csv', 'parquet', 'excel', 'json']
)

for fmt, success in results.items():
    print(f"{fmt}: {'OK' if success else 'FALHOU'}")

CLI

python main.py input.csv output.parquet \
    --source-type csv \
    --dest-type parquet \
    --remove-duplicates \
    --remove-nulls \
    --validate \
    --metrics

Docker

Build

docker-compose build

Executar

docker-compose up -d

Acessar Adminer

http://localhost:8080

Testes

Executar todos os testes

pytest tests/ -v

Com cobertura

pytest tests/ -v --cov=etl --cov-report=html

Testes específicos

pytest tests/test_integration.py -v

Estrutura do Projeto

.
├── etl/
│   ├── __init__.py
│   ├── connectors.py          # Conectores de dados
│   ├── database.py            # Conectores DB
│   ├── processors.py          # Processadores
│   ├── orchestrator.py        # Orquestrador
│   ├── pipeline.py            # Pipeline principal
│   ├── incremental.py         # Carga incremental
│   ├── quality.py             # Validação qualidade
│   ├── schema.py              # Validação schema
│   ├── parallel.py            # Processamento paralelo
│   ├── lineage.py             # Data lineage
│   ├── checkpoint.py          # Checkpoint/recovery
│   ├── export.py              # Exportação
│   └── batch_optimizer.py     # Otimização batch
├── tests/
│   ├── test_integration.py
│   ├── test_database.py
│   ├── test_quality.py
│   └── test_schema.py
├── main.py
├── requirements.txt
├── docker-compose.yml
├── Dockerfile
├── ARCHITECTURE.md
└── README.md

Tecnologias

• Python 3.8+
• pandas
• pytest
• PostgreSQL
• MySQL
• Docker
• pre-commit

Licença

MIT

Contato

Para dúvidas ou sugestões, abra uma issue.