AULA 01: A Trincheira da Ingestão — Unificando e Limpando Dados Multiformato com PySpark e Pandas

# ====================================================================
# AULA 1 - CÉLULA 1: PREPARANDO O AMBIENTE DE BIG DATA NO COLAB
# ====================================================================

# [COMANDO NOVO] !pip install: O ponto de exclamação (!) avisa ao Colab
# que este comando vai direto para o terminal Linux da máquina virtual.
# - pyspark: Motor de processamento distribuído para Big Data.
# - openpyxl: Permite ao Python ler e gravar planilhas Excel (.xlsx).
# - lxml: Motor rápido em linguagem C para ler árvores de tags (.xml).
# - -q (quiet): Instalação silenciosa para não poluir a tela da aula.
!pip install pyspark openpyxl lxml -q

# [COMANDO NOVO] import os / json: Bibliotecas nativas do Python para
# gerenciar arquivos do sistema operacional e estruturas no formato JSON.
import os
import json

# [COMANDO NOVO] import pandas as pd: Importa a biblioteca Pandas
# atribuindo o apelido universal 'pd' para manipular tabelas locais.
import pandas as pd

# [COMANDO NOVO] SparkSession: É o ponto de partida oficial (o motor)
# para criar e processar DataFrames distribuídos no Apache Spark.
from pyspark.sql import SparkSession

# [COMANDO NOVO] functions as F: Importa a caixa de ferramentas do
# PySpark (limpeza de texto, regex, condicionais) com o apelido 'F'.
from pyspark.sql import functions as F

# [COMANDO NOVO] DoubleType, StringType: Importa os tipos de dados
# oficiais do Spark para converter textos em números decimais com segurança.
from pyspark.sql.types import DoubleType, StringType

print("✅ Motor PySpark e bibliotecas instaladas com sucesso!")

# ====================================================================
# AULA 1 - CÉLULA 2A: GERANDO DADOS BRUTOS (CSV DO MKT E JSON DO APP)
# ====================================================================

# [COMANDO NOVO] Aspas Triplas ("""): Permitem criar um texto longo
# com várias linhas no Python preservando as quebras de linha (Enter).
# Erros propositais simulados no CSV:
# 1. Separador ponto e vírgula (;) típico de Excel em português;
# 2. Espaços inúteis no nome (" ana silva ") e e-mail em MAIÚSCULO;
# 3. Valores com símbolo "R$" e vírgula decimal;
# 4. Cliente Beatriz (103) sem e-mail cadastrado (vazio entre ;;);
# 5. Ana Silva duplicada na linha 104 com CPF sem pontos e "Sao Paulo" sem til!
csv_content = """id_cliente;nome_completo;email;documento;valor_gasto;cidade;feedback_sac
101; ana silva ;ANA@GMAIL.COM;111.222.333-44;R$ 1500,50;São Paulo;Adorei o atendimento, muito rápido!
102;Carlos Moura;carlos@hotmail.com;22233344455;R$ 300,00;Rio de Janeiro;Produto atrasou 10 dias, péssimo suporte.
103;Beatriz Costa;;333.444.555-66;R$ 0,00;Curitiba;Ainda não recebi meu acesso ao sistema.
104;ana silva;ana@gmail.com;11122233344;R$ 1500,50;Sao Paulo;Adorei o atendimento, muito rápido!"""

# [COMANDO NOVO] with open(..., "w", encoding="utf-8") as f:
# Cria um arquivo físico no Colab em modo de escrita ("w" = write).
# O parâmetro encoding="utf-8" protege acentos e 'ç'. O bloco 'with'
# fecha e salva o arquivo automaticamente ao terminar o bloco.
with open("clientes_marketing.csv", "w", encoding="utf-8") as f:
    # [COMANDO NOVO] .strip(): Remove espaços ou quebras de linha
    # vazias que tenham sobrado nas extremidades do texto.
    f.write(csv_content.strip())

# Criando a estrutura do Aplicativo Mobile (Lista de Dicionários JSON).
# Repare que os nomes das colunas estão em inglês e há um valor None (nulo).
json_data = [
    {
        "user_id": "201",
        "full_name": "Marcos Vinicius",
        "contact_mail": "marcos@tech.com",
        "cpf": "444.555.666-77",
        "total_spent": 4520.00,
        "location": "Belo Horizonte",
        "review": "O app trava na tela de pagamento, quase desisti."
    },
    {
        "user_id": "202",
        "full_name": "  FERNANDA LIMA ",
        "contact_mail": "fernanda@lima.com.br",
        "cpf": "55566677788",
        "total_spent": None,
        "location": "Porto Alegre",
        "review": "Cancelei minha assinatura, achei caro pelo que entrega."
    },
    {
        "user_id": "203",
        "full_name": "Roberto Dias",
        "contact_mail": "roberto@dias.com",
        "cpf": "666.777.888-99",
        "total_spent": 890.90,
        "location": "São Paulo",
        "review": "Excelente custo-benefício, recomendo para todos!"
    }
]

# [COMANDO NOVO] json.dump(..., ensure_ascii=False, indent=4):
# Grava a lista do Python como um arquivo JSON oficial no disco:
# - ensure_ascii=False: Mantém os acentos reais no arquivo.
# - indent=4: Formata o código com 4 espaços de recuo para leitura humana.
with open("clientes_app.json", "w", encoding="utf-8") as f:
    json.dump(json_data, f, ensure_ascii=False, indent=4)

print("📁 Parte 1 concluída: 'clientes_marketing.csv' e 'clientes_app.json' criados!")

# ====================================================================
# AULA 1 - CÉLULA 2B: GERANDO DADOS BRUTOS (EXCEL E XML DO ERP LEGADO)
# ====================================================================

# [COMANDO NOVO] pd.DataFrame({...}): Cria uma tabela estruturada em
# memória usando o Pandas. Cada chave do dicionário vira uma coluna.
# Erros propositais: Texto "N/A" na coluna numérica de faturamento e
# a cliente Ana Silva (CPF 111.222.333-44) repetida mais uma vez!
df_excel = pd.DataFrame({
    "cod_cli": [301, 302, 303],
    "cliente": ["Juliana Paes", "Pedro Alvares", "Ana Silva"],
    "e_mail": ["ju@paes.com", "pedro@brasil.com", "ana@gmail.com"],
    "nr_cpf": ["777.888.999-00", "888.999.000-11", "111.222.333-44"],
    "faturamento": ["2100.75", "N/A", "1500.50"],
    "uf_cidade": ["Salvador", "Recife", "São Paulo"],
    "observacao": [
        "Cliente VIP, sempre paga em dia.",
        "Pediu desconto na renovação, ameaçou sair.",
        "Comprou pacote anual."
    ]
})

# [COMANDO NOVO] .to_excel(..., index=False): Salva a tabela do Pandas
# diretamente como uma planilha oficial do Microsoft Excel (.xlsx).
# O argumento index=False impede a criação de uma coluna extra (0, 1, 2).
df_excel.to_excel("clientes_financeiro.xlsx", index=False)

# Criando um arquivo XML (típico de sistemas de Notas Fiscais e ERPs antigos).
# No XML, os dados ficam envoltos em tags de abertura <tag> e fechamento </tag>.
xml_content = """<?xml version="1.0" encoding="UTF-8"?>
<cadastros>
    <registro>
        <codigo>401</codigo>
        <nome>Camila Pitanga</nome>
        <correio>camila@globo.com</correio>
        <cadastro_pessoa>999.888.777-66</cadastro_pessoa>
        <receita>3200.00</receita>
        <municipio>Rio de Janeiro</municipio>
        <historico>Renovou contrato ontem, muito elogiou a equipe.</historico>
    </registro>
    <registro>
        <codigo>402</codigo>
        <nome>  bruno gagliasso</nome>
        <correio>BRUNO@SURUBIM.COM</correio>
        <cadastro_pessoa>12345678909</cadastro_pessoa>
        <receita>150.00</receita>
        <municipio>Sao Paulo</municipio>
        <historico>Furioso com cobrança indevida no cartão.</historico>
    </registro>
</cadastros>"""

with open("clientes_erp_legado.xml", "w", encoding="utf-8") as f:
    f.write(xml_content.strip())

print("📁 Parte 2 concluída: Todos os 4 arquivos brutos estão prontos na pasta do Colab!")

# ====================================================================
# AULA 1 - CÉLULA 3A: INGESTÃO NATIVA NO PYSPARK (CSV E JSON)
# ====================================================================

# [COMANDO NOVO] SparkSession.builder.appName(...).getOrCreate():
# Inicia o motor do Spark na memória. Se já existir uma sessão aberta,
# ele a reaproveita; caso contrário, cria uma nova na hora.
spark = (
    SparkSession.builder
    .appName("Aula1_EngenhariaDados")
    .getOrCreate()
)

# [COMANDO NOVO] spark.read.option(...).csv():
# Lê o arquivo CSV diretamente no cluster do Spark:
# - "header", "true": Usa a primeira linha como cabeçalho das colunas.
# - "delimiter", ";": Informa que o separador é ponto e vírgula.
df_csv_bruto = (
    spark.read
    .option("header", "true")
    .option("delimiter", ";")
    .csv("clientes_marketing.csv")
)

# [COMANDO NOVO] .select(), F.col(), .alias() e F.lit():
# - .select(): Escolhe e ordena as colunas da tabela.
# - F.col("antigo").alias("novo"): Renomeia a coluna para o padrão único.
# - F.lit("CSV_Marketing"): "lit" significa Valor Literal (constante).
#   Cria uma coluna nova com o texto fixo "CSV_Marketing" em todas as
#   linhas para rastrearmos de qual sistema o cliente veio!
df_csv_padrao = df_csv_bruto.select(
    F.col("id_cliente").alias("id_origem"),
    F.col("nome_completo").alias("nome"),
    F.col("email").alias("email"),
    F.col("documento").alias("cpf"),
    F.col("valor_gasto").alias("valor"),
    F.col("cidade").alias("cidade"),
    F.col("feedback_sac").alias("comentario"),
    F.lit("CSV_Marketing").alias("origem_dados")
)

# [COMANDO NOVO] .option("multiline", "true").json():
# Obrigatório quando o arquivo JSON está formatado em múltiplas linhas
# dentro de uma lista [ ]. Sem isso, o Spark dá erro de leitura!
df_json_bruto = (
    spark.read
    .option("multiline", "true")
    .json("clientes_app.json")
)

# [COMANDO NOVO] .cast(StringType()):
# Converte temporariamente o número de 'total_spent' para Texto (String),
# permitindo empilhar com os outros arquivos que ainda contêm "R$" ou "N/A".
df_json_padrao = df_json_bruto.select(
    F.col("user_id").alias("id_origem"),
    F.col("full_name").alias("nome"),
    F.col("contact_mail").alias("email"),
    F.col("cpf").alias("cpf"),
    F.col("total_spent").cast(StringType()).alias("valor"),
    F.col("location").alias("cidade"),
    F.col("review").alias("comentario"),
    F.lit("JSON_App").alias("origem_dados")
)

print("✅ Arquivos CSV e JSON alinhados para o Schema Padrão!")

# ====================================================================
# AULA 1 - CÉLULA 3B: INGESTÃO EXCEL/XML E EMPILHAMENTO (UNIONBYNAME)
# ====================================================================

# [COMANDO NOVO] pd.read_excel(..., dtype=str) + spark.createDataFrame():
# Pulo do gato do mercado: Para arquivos Excel e XML, lemos primeiro com
# Pandas forçando tudo como texto (dtype=str) para não perder zeros à
# esquerda no CPF, e convertemos para o Spark com createDataFrame()!
pdf_excel = pd.read_excel("clientes_financeiro.xlsx", dtype=str)

df_excel_padrao = spark.createDataFrame(pdf_excel).select(
    F.col("cod_cli").alias("id_origem"),
    F.col("cliente").alias("nome"),
    F.col("e_mail").alias("email"),
    F.col("nr_cpf").alias("cpf"),
    F.col("faturamento").alias("valor"),
    F.col("uf_cidade").alias("cidade"),
    F.col("observacao").alias("comentario"),
    F.lit("XLSX_Financeiro").alias("origem_dados")
)

# [COMANDO NOVO] pd.read_xml(..., dtype=str):
# Lê a árvore de tags XML e já transforma cada <registro> em uma linha.
pdf_xml = pd.read_xml("clientes_erp_legado.xml", dtype=str)

df_xml_padrao = spark.createDataFrame(pdf_xml).select(
    F.col("codigo").alias("id_origem"),
    F.col("nome").alias("nome"),
    F.col("correio").alias("email"),
    F.col("cadastro_pessoa").alias("cpf"),
    F.col("receita").alias("valor"),
    F.col("municipio").alias("cidade"),
    F.col("historico").alias("comentario"),
    F.lit("XML_ERP").alias("origem_dados")
)

# [COMANDO NOVO] .unionByName():
# Empilha as 4 tabelas verticalmente casando pelo NOME exato da coluna
# (muito mais seguro que o .union() comum, que casa apenas pela posição).
df_bronze = (
    df_csv_padrao
    .unionByName(df_json_padrao)
    .unionByName(df_excel_padrao)
    .unionByName(df_xml_padrao)
)

# [COMANDO NOVO] .count() e .show(truncate=False):
# - .count(): Conta o total de registros da tabela (12 linhas brutas).
# - .show(truncate=False): Exibe a tabela inteira sem cortar textos longos.
print(f"📦 Total de registros brutos na Camada Bronze: {df_bronze.count()}")
df_bronze.show(truncate=False)

# ====================================================================
# AULA 1 - CÉLULA 4A: DATA CLEANING COM REGEX E CONDICIONAIS NO SPARK
# ====================================================================

# [COMANDO NOVO] .withColumn("nome_coluna", regra_de_transformacao):
# Cria uma coluna nova ou substitui uma coluna existente aplicando
# uma função em todas as linhas paralelamente.
df_limpo = (
    df_bronze
    
    # [COMANDO NOVO] F.trim() e F.initcap():
    # - F.trim(): Remove espaços em branco no início e no fim do texto.
    # - F.initcap(): Deixa a 1ª letra de cada nome Maiúscula ("Ana Silva").
    .withColumn("nome", F.initcap(F.trim(F.col("nome"))))
    
    # [COMANDO NOVO] F.lower() e F.coalesce():
    # - F.lower(): Converte todo o e-mail para letras minúsculas.
    # - F.coalesce(col, padrao): Se o e-mail estiver vazio/nulo (None),
    #   preenche automaticamente com "sem_email@cadastro.com".
    .withColumn(
        "email",
        F.coalesce(
            F.lower(F.trim(F.col("email"))),
            F.lit("sem_email@cadastro.com")
        )
    )
    
    # [COMANDO NOVO] F.regexp_replace(coluna, padrao_regex, substituto):
    # A Expressão Regular "[^0-9]" significa: "Tudo que NÃO (^) for dígito
    # numérico de 0 a 9". Apaga pontos, traços e espaços do CPF na hora!
    .withColumn("cpf", F.regexp_replace(F.col("cpf"), "[^0-9]", ""))
    
    # Remove o símbolo "R$" e espaços em branco (\s) da coluna de valor:
    .withColumn("v_limpo", F.regexp_replace(F.col("valor"), "[R$\\s]", ""))
    
    # Troca a vírgula brasileira pelo ponto decimal americano:
    .withColumn("v_limpo", F.regexp_replace(F.col("v_limpo"), ",", "."))
    
    # [COMANDO NOVO] F.when(condicao, valor_se_sim).otherwise(valor_se_nao):
    # É o "IF / ELSE" do PySpark!
    # [COMANDO NOVO] .rlike("^[0-9]+(\\.[0-9]+)?$"): Testa via Regex se o
    # texto contém apenas números válidos. Se sim, converte para DoubleType
    # (decimal); se for "N/A" ou nulo, preenche com 0.0 sem dar erro!
    .withColumn(
        "valor_gasto",
        F.when(
            F.col("v_limpo").rlike("^[0-9]+(\\.[0-9]+)?$"),
            F.col("v_limpo").cast(DoubleType())
        ).otherwise(0.0)
    )
    
    # Corrige "Sao Paulo" sem til para "São Paulo" (evita duplicar cidade):
    .withColumn(
        "cidade",
        F.regexp_replace(F.trim(F.col("cidade")), "^Sao Paulo$", "São Paulo")
    )
    
    # [COMANDO NOVO] .drop(): Descarta as colunas auxiliares antigas.
    .drop("valor", "v_limpo")
)

print("🧹 Regras de limpeza e conversão numérica aplicadas!")

# ====================================================================
# AULA 1 - CÉLULA 4B: DEDUPLICAÇÃO E SALVAMENTO DO ENTREGÁVEL SILVER
# ====================================================================

# [COMANDO NOVO] .dropDuplicates(["cpf"]):
# Inspeciona a coluna 'cpf' (agora limpa, contendo apenas 11 números)
# e elimina qualquer linha repetida, mantendo apenas 1 registro único
# por cliente (Golden Record). As 2 cópias extras da Ana Silva somem aqui!
df_silver = df_limpo.dropDuplicates(["cpf"])

print(f"✨ Total de clientes únicos na Camada Silver: {df_silver.count()}")
df_silver.show(truncate=False)

# [COMANDO NOVO] .toPandas().to_csv(..., index=False):
# Quando o Spark salva arquivos nativamente, ele cria uma pasta com
# vários arquivos fragmentados (part-0000...). Convertendo o resultado
# limpo para Pandas com .toPandas(), geramos um arquivo CSV único
# perfeito que será a base de entrada da nossa Aula 2!
df_silver.toPandas().to_csv(
    "base_silver_aula1.csv",
    index=False,
    encoding="utf-8"
)

print("💾 Entregável da Aula 1 salvo com sucesso: 'base_silver_aula1.csv'!")