Netflix Analytics Pipeline

Pipeline de dados end-to-end para análise de uma plataforma de streaming fictícia com 465 mil assinantes e 5 milhões de eventos de visualização. Desenvolvido como projeto de portfólio enterprise com PySpark, Delta Lake, Databricks e CrewAI.

---

Arquitetura

Dados Sintéticos (5M+ eventos)
         │
         ▼
    ┌─────────┐
    │  BRONZE │  Ingestão append-only com lineage (_ingested_at, _source_file, _batch_id)
    │  Delta  │  Particionado por data_evento
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  SILVER │  Sessionização via window functions · Validação · DLQ (rejeições)
    │  Delta  │  MERGE idempotente · Quality Report · MAX_REJECTION_RATE=0.25
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  GOLD   │  6 marts analíticos: engagement, conteúdo, LTV, churn, receita, binge
    │ Parquet │  Broadcast joins · AQE · Window functions para rankings e churn
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │ DIAMOND │  4 marts executivos no Databricks SQL Warehouse
    │   SQL   │  dm_executive_kpis · dm_content_top10 · dm_churn_cohort · dm_revenue_forecast
    └────┬────┘
         │
         ▼
    ┌─────────┐
    │  CrewAI │  2 workflows multi-agente via Claude Sonnet
    │ Agentes │  Negócio (5 agentes) · Engenharia (5 agentes)
    └─────────┘

Stack

| Camada | Tecnologia | |--------|-----------| | Processamento | PySpark 4.x + Delta Lake 3.x | | Armazenamento local | Delta / Parquet em data/output/ | | Lakehouse | Databricks SQL Warehouse | | Governança | Unity Catalog (comments + tags) | | Agentes IA | CrewAI + Claude Sonnet (Anthropic) | | Testes | pytest + pytest-cov (88 testes) | | CI/CD | GitHub Actions (test + lint + type-check) | | Qualidade | Ruff + Pyright |

---

Estrutura do Projeto

netflix-pipeline/
├── config/
│   ├── settings.py          # Configurações por ambiente (dev/databricks)
│   └── schemas.py           # Schemas Spark das camadas Bronze/Silver/Gold
│
├── data/
│   ├── generate_data.py     # Gerador: 5M+ eventos com problemas de qualidade
│   ├── dicionario_dados.json # Referência anti-alucinação para agentes CrewAI
│   └── output/              # Artefatos Delta/Parquet gerados localmente
│
├── pipeline/
│   ├── ingest.py            # Bronze: ingestão append-only com lineage
│   ├── quality.py           # Silver: sessionização, validação e DLQ
│   ├── transform.py         # Gold: 6 marts com métricas Netflix
│   ├── delta_utils.py       # Utilitários Delta Lake (MERGE, vacuum, optimize)
│   └── watermark.py         # Controle de watermark para ingestão idempotente
│
├── databricks/
│   ├── client.py            # Cliente REST API Databricks
│   ├── upload_gold.py       # Upload Gold → SQL Warehouse (INSERT/MERGE)
│   ├── create_diamond_layer.py # Camada Diamond (4 marts executivos)
│   ├── jobs.py              # Jobs API 2.1 (CRUD + polling)
│   ├── setup_jobs.py        # Job diário às 06:00 BRT + notebook Diamond
│   ├── setup_secrets.py     # Scope 'netflix-pipeline' + secrets
│   ├── setup_monitoring.py  # Health check notebook + alertas por email
│   ├── setup_dashboard.py   # Dashboard matplotlib 5 seções
│   ├── setup_unity_catalog.py # Comments + tags em 10 tabelas
│   ├── setup_all.py         # Orquestrador de setup completo
│   └── run_full_pipeline.py # Pipeline completo local → Databricks
│
├── crew/
│   ├── ferramentas.py       # 6 ferramentas analíticas (leitura Gold via pandas)
│   ├── agentes.py           # 5 agentes de negócio
│   ├── tarefas.py           # 5 tarefas do workflow de negócio
│   ├── ferramentas_engenharia.py # 7 ferramentas técnicas (schemas, código, perfil)
│   ├── agentes_engenharia.py    # 5 agentes de engenharia
│   └── tarefas_engenharia.py   # 5 tarefas do workflow de engenharia
│
├── tests/
│   ├── conftest.py          # SparkSession compartilhada (scope=session)
│   ├── test_quality.py      # Testes da camada Silver (sessionização, DLQ)
│   └── test_transform.py    # Testes da camada Gold (métricas, churn, binge)
│
├── .github/workflows/
│   └── ci.yml               # GitHub Actions: test + lint + type-check
│
├── main.py                  # Entry point local: Bronze → Silver → Gold
├── run_crew.py              # Entry point CrewAI — workflow de negócio
├── run_crew_engenharia.py   # Entry point CrewAI — workflow de engenharia
├── requirements.txt
├── .env.dev                 # Variáveis de ambiente (dev local)
└── .env.databricks          # Variáveis de ambiente (Databricks)

---

Pré-requisitos

| Requisito | Versão | |-----------|--------| | Python | 3.12+ | | Java | 17 (requerido pelo PySpark 4.x) | | Git | qualquer |

Windows: necessário HADOOP_HOME apontando para o diretório com winutils.exe.

---

Setup Local

1. Clonar e instalar

git clone <url-do-repo>
cd netflix-pipeline
pip install -r requirements.txt

2. Configurar variáveis de ambiente

Edite .env.dev com os valores do seu ambiente:

ENV=dev
OUTPUT_PATH=data/output
LOG_LEVEL=DEBUG
SHUFFLE_PARTITIONS=4

# Necessário apenas para run_crew.py
ANTHROPIC_API_KEY=sk-ant-...

3. Gerar dados sintéticos

python data/generate_data.py

Gera ~5 milhões de eventos em data/raw/ com problemas de qualidade intencionais (FK inválidas, campos nulos, timestamps fora de ordem).

4. Executar o pipeline

python main.py

Processa Bronze → Silver → Gold e escreve os artefatos em data/output/. Tempo estimado: 3-8 minutos dependendo da máquina.

5. Executar os testes

python -m pytest tests/ -v

88 testes, tempo estimado: ~20 segundos.

---

Tabelas Gold

Geradas localmente em data/output/gold/ após python main.py.

| Tabela | Descrição | Grão | |--------|-----------|------| | engagement_daily | DAU, watch hours, completion rate, binge sessions | dia | | content_performance | Viewers, completion, skip rate, ranking por gênero | título | | user_ltv | LTV estimado, plano, país, segmento de valor | usuário | | churn_indicators | Risco (LOW/MEDIUM/HIGH), dias inativo, queda de sessões | usuário | | revenue_metrics | MRR, ARR, churn rate, novos vs cancelados por plano | mês × plano | | binge_sessions | Sessões binge, episódios assistidos, tendência | título × mês |

---

Tabelas Diamond (Databricks)

Criadas no SQL Warehouse após python databricks/run_full_pipeline.py.

| Tabela | Descrição | |--------|-----------| | dm_executive_kpis | KPIs consolidados: DAU médio, completion rate, churn rate, MRR | | dm_content_top10 | Top 10 títulos por score de qualidade (viewers × completion × binge) | | dm_churn_cohort | Distribuição de risco de churn por plano e faixa de inatividade | | dm_revenue_forecast | Projeção de MRR para os próximos 3 meses (pessimista e otimista) |

---

Integração Databricks

Pré-requisitos

• Workspace Databricks com SQL Warehouse ativo
• Token de acesso pessoal

Configuração

Preencha .env.databricks:

DATABRICKS_HOST=https://adb-xxxx.azuredatabricks.net
DATABRICKS_TOKEN=dapi...
DATABRICKS_WAREHOUSE_ID=xxxx

Setup do workspace (primeira vez)

python databricks/setup_all.py [email protected]

Configura em sequência: job diário, dashboard, health check, secrets e Unity Catalog.

Executar pipeline completo

# Pipeline local + upload Gold + Diamond
python databricks/run_full_pipeline.py

# Com trigger do job agendado
python databricks/run_full_pipeline.py --trigger

# Apenas upload (Gold já gerado)
python databricks/run_full_pipeline.py --skip-local

---

Agentes CrewAI

Dois workflows independentes de análise via LLM.

Workflow de Negócio

Cinco agentes sequenciais que analisam os dados Gold e geram um relatório executivo.

pipeline_ops → content_analyst → churn_detector → revenue_analyst → vp_strategy

# Requer ANTHROPIC_API_KEY no .env.dev
python run_crew.py --verbose

Saída: data/output/relatorio_executivo.md

Workflow de Engenharia

Cinco agentes que auditam o código e a qualidade dos dados e geram um relatório técnico.

contextualizador → analista_qualidade → engenheiro_pipeline → validador_dados → relator_executivo

python run_crew_engenharia.py --verbose

Saída: data/output/relatorio_engenharia.md

---

CI/CD

GitHub Actions em .github/workflows/ci.yml com três jobs:

| Job | Trigger | O que valida | |-----|---------|-------------| | test | push/PR | pytest 88 testes + cobertura | | lint | push/PR | Ruff (E, F, W, I) | | type-check | push/PR | Pyright em pipeline/ e config/ |

Dispara em push para main/develop e em pull requests para main.

---

Diferenciais Netflix vs E-commerce

| Aspecto | Detalhe | |---------|---------| | Sessionização | Reconstrução de sessões via window functions (sem eventos de sessão explícitos) | | Churn sem ML | Detecção de churn via raciocínio do agente sobre dados comportamentais | | Binge-watching | Métrica proprietária: sessões com ≥ 2 episódios consecutivos | | Completion rate | Watch time / content duration — indicador chave de retenção | | DLQ tolerante | MAX_REJECTION_RATE=0.25 (cascata FK em dados sintéticos justifica tolerância maior) | | Agentes duplos | Workflow de negócio + workflow de engenharia independentes |

---

Licença

Projeto educacional e de portfólio. Dados 100% sintéticos — nenhuma informação real de usuários foi utilizada.