1. Visão Geral do Caso de Negócio & Arquitetura Medallion
Contextualização • 00–15 min
Na rotina real de Engenharia e Ciência de Dados, as informações nunca estão centralizadas em uma única planilha limpa. Nesta aula, simulamos uma empresa que adquiriu três concorrentes e precisa unificar imediatamente as bases de clientes de quatro setores: Marketing (CSV), Aplicativo Mobile (JSON), Financeiro (Excel XLSX) e ERP Antigo (XML). Construiremos a transição da Camada Bronze (ingestão bruta empilhada) para a Camada Silver (dados padronizados, tipados e sem CPFs duplicados).
Arquitetura Medallion da Aula 01 — Fluxo de Dados Brutos até a Base Silver
• Marketing (.csv)
• App Mobile (.json)
• Financeiro (.xlsx)
• ERP Fiscal (.xml)
Dados Sujos / Caos
➔
Schema Alignment
.select() + .alias()
Origem via F.lit()
.unionByName()
12 Linhas Empilhadas
➔
F.trim() / F.initcap()
Regex CPF e Moeda
F.when().otherwise()
.dropDuplicates()
Faxina Distribuída
➔
CPFs de 11 Dígitos
Valores em Double
Sem Duplicatas
Pronto p/ Aula 02 (IA)
10 Clientes Únicos
| Janela de Tempo |
Etapa Pedagógica |
Problema Real Simulado em Aula |
Entregável Parcial no Colab |
| 00 – 15 min |
Setup e Dependências (Célula 1) |
O ambiente padrão não possui o motor Apache Spark nem leitores nativos de Excel e XML instalados. |
Sessão PySpark e bibliotecas prontas na memória. |
| 15 – 40 min |
Fábrica de Caos (Células 2A e 2B) |
Arquivos com separadores distintos, valores com R$, textos "N/A", nulos e a mesma cliente (Ana Silva) repetida 3 vezes. |
4 arquivos físicos (.csv, .json, .xlsx, .xml) gerados no Colab. |
| 40 – 65 min |
Ingestão e Schema Alignment (Células 3A e 3B) |
Colunas com nomes divergentes entre setores (documento vs cpf vs nr_cpf vs cadastro_pessoa). |
DataFrame df_bruto_unificado com 12 registros alinhados. |
| 65 – 90 min |
Faxina Pesada e Deduplicação (Célula 4) |
CPFs com pontos/traços impedem achar duplicatas; cidades sem acento (Sao Paulo) dividem gráficos. |
Arquivo único higienizado: base_silver_aula1.csv (10 clientes). |
2. Célula 1 — Preparação do Ambiente e Instalação de Dependências
00–15 min
Instrua os alunos a abrirem um novo notebook no Google Colab, colarem o código abaixo na primeira célula e executarem com Shift + Enter. Todo comando inédito possui a marcação didática # [COMANDO NOVO]:
# ====================================================================
# AULA 1 - CÉLULA 1: PREPARAÇÃO DO AMBIENTE E MOTOR DE BIG DATA
# ====================================================================
# [COMANDO NOVO] !pip install:
# O ponto de exclamação (!) executa um comando direto no terminal Linux
# da máquina virtual do Google Colab:
# - pyspark: Motor de processamento distribuído em larga escala.
# - openpyxl: Driver obrigatório para ler e gravar planilhas .xlsx.
# - lxml: Motor rápido para ler estruturas de tags hierárquicas .xml.
# - -q (quiet): Modo silencioso (evita centenas de linhas na tela).
!pip install pyspark openpyxl lxml -q
# [COMANDO NOVO] import os / json:
# Bibliotecas nativas do Python para gerenciar arquivos do sistema
# operacional e manipular objetos estruturados em formato JSON.
import os
import json
# [COMANDO NOVO] import pandas as pd:
# Carrega a biblioteca Pandas com o apelido mundial 'pd' para
# manipulação rápida de tabelas em memória local.
import pandas as pd
# [COMANDO NOVO] SparkSession:
# É o objeto central que inicializa e controla o cluster do PySpark.
from pyspark.sql import SparkSession
# [COMANDO NOVO] functions as F:
# Importa a biblioteca de funções nativas do Spark (limpeza de texto,
# regex, condicionais) com o apelido curto 'F'.
from pyspark.sql import functions as F
# [COMANDO NOVO] DoubleType, StringType:
# Tipos oficiais de colunas do Spark para forçar conversões seguras
# entre Texto (String) e Número Decimal (Double).
from pyspark.sql.types import DoubleType, StringType
print("✅ Ambiente configurado e bibliotecas carregadas!")
3. Células 2A e 2B — A Fábrica de Caos (Gerando os 4 Arquivos Sujos)
15–40 min
💡 Dinâmica de Sala de Aula (Zero Atrito com Arquivos Externos)
Em vez de perder tempo enviando arquivos avulsos por e-mail ou Drive para os alunos subirem no Colab, as Células 2A e 2B usam o próprio Python para fabricar os 4 arquivos brutos com todas as anomalias reais de mercado. Peça aos alunos que abram a aba lateral esquerda do Colab (ícone de pasta 📁) para inspecionar os arquivos criados.
# ====================================================================
# AULA 1 - CÉLULA 2A: GERANDO DADOS BRUTOS (CSV DO MKT E JSON DO APP)
# ====================================================================
# [COMANDO NOVO] Aspas Triplas ("""):
# Permitem criar strings de múltiplas linhas no Python mantendo quebras.
# Erros reais simulados neste CSV:
# - Separador ponto e vírgula (;), valores com "R$" e vírgula decimal;
# - Espaços extras (" ana silva "), e-mail maiúsculo e e-mail vazio (103);
# - Ana Silva duplicada na linha 104 com CPF sem pontos e "Sao Paulo".
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 o arquivo físico no disco no modo escrita ("w" = write) com
# padrão UTF-8 (preserva acentos) e fecha o arquivo automaticamente.
with open("clientes_marketing.csv", "w", encoding="utf-8") as f:
# [COMANDO NOVO] .strip():
# Remove espaços ou quebras de linha vazias nas pontas do texto.
f.write(csv_content.strip())
# Criando os dados do App Mobile (Lista de dicionários chave:valor).
# Note as chaves em inglês e o gasto nulo (None) na cliente Fernanda.
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."
},
{
"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):
# Exporta a lista Python para um arquivo .json formatado com 4 espaços
# de recuo (indent=4) e mantendo acentuação normal (ensure_ascii=False).
with open("clientes_app.json", "w", encoding="utf-8") as f:
json.dump(json_data, f, ensure_ascii=False, indent=4)
print("📁 Arquivos CSV e JSON gerados com sucesso!")
# ====================================================================
# AULA 1 - CÉLULA 2B: GERANDO DADOS BRUTOS (EXCEL E XML DO ERP)
# ====================================================================
# [COMANDO NOVO] pd.DataFrame({...}):
# Constrói uma tabela em memória no Pandas a partir de um dicionário.
# Erros simulados: texto "N/A" na coluna de faturamento e o CPF
# 111.222.333-44 (Ana Silva) repetido também na base do Financeiro!
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):
# Grava o DataFrame em um arquivo Excel (.xlsx). O parâmetro index=False
# impede que o Pandas salve uma coluna extra sem nome com (0, 1, 2).
df_excel.to_excel("clientes_financeiro.xlsx", index=False)
# Criando o arquivo XML legado (típico de sistemas fiscais e ERPs).
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, 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("📁 Arquivos XLSX e XML gerados! Todos os 4 formatos prontos.")
4. Células 3A e 3B — Ingestão no PySpark e Alinhamento de Schema
40–65 min
Com os quatro arquivos no disco, iniciamos a ingestão distribuída. Para empilhar tabelas de sistemas diferentes, aplicamos o Schema Alignment: traduzimos os nomes originais de cada coluna para um padrão único (id_origem, nome, email, cpf, valor, cidade, comentario e origem_dados).
# ====================================================================
# AULA 1 - CÉLULA 3A: INGESTÃO NO PYSPARK (CSV E JSON)
# ====================================================================
# [COMANDO NOVO] SparkSession.builder.appName(...).getOrCreate():
# Liga o motor do PySpark no Colab (ou recupera a sessão já ativa).
spark = (
SparkSession.builder
.appName("Aula1_EngenhariaDados")
.getOrCreate()
)
# [COMANDO NOVO] spark.read.option(...).csv():
# Configura o leitor de CSV do Spark informando que a 1ª linha é o
# cabeçalho ("header", "true") e que o separador é ";" ("delimiter").
df_csv_bruto = (
spark.read
.option("header", "true")
.option("delimiter", ";")
.csv("clientes_marketing.csv")
)
# [COMANDO NOVO] .select(), F.col(), .alias() e F.lit():
# - F.col("antigo").alias("novo"): Renomeia cada coluna para o padrão.
# - F.lit("CSV_Marketing"): Cria uma coluna constante (Literal) para
# registrar de qual sistema legado aquele registro 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():
# Permite ao Spark ler arquivos JSON indentados em múltiplas linhas.
df_json_bruto = (
spark.read
.option("multiline", "true")
.json("clientes_app.json")
)
# [COMANDO NOVO] .cast(StringType()):
# Converte 'total_spent' para texto temporariamente para permitir o
# empilhamento com os 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("✅ Bases CSV e JSON lidas e padronizadas no PySpark!")
# ====================================================================
# AULA 1 - CÉLULA 3B: INGESTÃO DE EXCEL/XML E UNIÃO DAS 4 TABELAS
# ====================================================================
# [COMANDO NOVO] pd.read_excel(..., dtype=str):
# Lê a planilha Excel forçando todas as colunas como texto (str) para
# evitar que zeros à esquerda no CPF sejam apagados.
pdf_excel = pd.read_excel("clientes_financeiro.xlsx", dtype=str)
# [COMANDO NOVO] spark.createDataFrame():
# Transfere a tabela do Pandas direto para o cluster distribuído Spark!
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):
# Interpreta as tags XML do ERP antigo e transforma em tabela.
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 conectando cada coluna pelo NOME exato!
df_bruto_unificado = (
df_csv_padrao
.unionByName(df_json_padrao)
.unionByName(df_excel_padrao)
.unionByName(df_xml_padrao)
)
# [COMANDO NOVO] .count() e .show(truncate=False):
# Conta as linhas (12) e exibe a tabela completa sem cortar textos.
total_bruto = df_bruto_unificado.count()
print(f"📦 Total de linhas brutas empilhadas: {total_bruto}")
df_bruto_unificado.show(truncate=False)
5. Célula 4 — Faxina Pesada (Data Cleaning), Deduplicação e Base Silver
65–90 min
Na etapa final da Aula 01, aplicamos transformações em lote com Expressões Regulares (Regex) para limpar CPFs e valores monetários, tratamos textos inválidos ("N/A") com condicionais distribuídas e eliminamos os cadastros duplicados antes de exportar o arquivo base_silver_aula1.csv:
# ====================================================================
# AULA 1 - CÉLULA 4: DATA CLEANING E GERAÇÃO DA CAMADA SILVER
# ====================================================================
# [COMANDO NOVO] .withColumn("coluna", regra):
# Cria uma coluna nova ou atualiza uma coluna existente na tabela.
df_limpo = (
df_bruto_unificado
# [COMANDO NOVO] F.trim() e F.initcap():
# Remove espaços nas pontas e deixa Iniciais Maiúsculas ("Ana Silva").
.withColumn("nome", F.initcap(F.trim(F.col("nome"))))
# [COMANDO NOVO] F.lower() e F.coalesce():
# Passa o e-mail para minúsculas e, se for nulo, preenche valor padrão.
.withColumn(
"email",
F.coalesce(
F.lower(F.trim(F.col("email"))),
F.lit("sem_email@cadastro.com")
)
)
# [COMANDO NOVO] F.regexp_replace(col, "[^0-9]", ""):
# Regex que apaga tudo que NÃO for número de 0 a 9 no CPF!
.withColumn("cpf", F.regexp_replace(F.col("cpf"), "[^0-9]", ""))
# Limpa "R$" e espaços, e depois troca a vírgula por ponto decimal:
.withColumn(
"valor_limpo",
F.regexp_replace(F.col("valor"), "[R$\\s]", "")
)
.withColumn(
"valor_limpo",
F.regexp_replace(F.col("valor_limpo"), ",", ".")
)
# [COMANDO NOVO] F.when(condicao, sim).otherwise(nao) + .rlike():
# Testa via Regex (.rlike) se o valor é um número válido. Se for,
# converte para DoubleType; se for "N/A" ou nulo, preenche 0.0!
.withColumn(
"valor_gasto",
F.when(
F.col("valor_limpo").rlike("^[0-9]+(\\.[0-9]+)?$"),
F.col("valor_limpo").cast(DoubleType())
).otherwise(0.0)
)
# Padroniza "Sao Paulo" sem acento para "São Paulo":
.withColumn(
"cidade",
F.regexp_replace(
F.trim(F.col("cidade")), "^Sao Paulo$", "São Paulo"
)
)
# [COMANDO NOVO] .drop():
# Remove as colunas temporárias de texto que já foram tratadas.
.drop("valor", "valor_limpo")
)
# [COMANDO NOVO] .dropDuplicates(["cpf"]):
# Garante apenas 1 cliente único por CPF! As duplicatas da Ana Silva
# que estavam espalhadas no CSV e no Excel são removidas aqui.
df_silver = df_limpo.dropDuplicates(["cpf"])
total_limpo = df_silver.count()
print(f"✨ Total de clientes únicos (Camada Silver): {total_limpo}")
df_silver.show(truncate=False)
# [COMANDO NOVO] .toPandas().to_csv():
# Exporta a base higienizada como um arquivo CSV único para a Aula 2!
df_silver.toPandas().to_csv(
"base_silver_aula1.csv", index=False, encoding="utf-8"
)
print("💾 Entregável salvo com sucesso: 'base_silver_aula1.csv'!")
6. Dicionário Técnico (Cheat Sheet) & Desafio de Fixação
15 min finais
| Comando / Função |
Papel na Engenharia de Dados |
Exemplo Prático na Aula |
F.col("x").alias("y") |
Seleciona a coluna original e padroniza seu nome (Schema Alignment). |
documento ➔ cpf |
F.lit("origem") |
Insere um valor constante em todas as linhas para auditoria de origem. |
Preenche "CSV_Marketing" |
df1.unionByName(df2) |
Empilha tabelas conectando colunas pelo nome (evita erros de posição). |
Une CSV + JSON + XLSX + XML |
F.initcap(F.trim(col)) |
Remove espaços laterais inúteis e capitaliza as iniciais de cada nome. |
" ana silva " ➔ "Ana Silva" |
F.regexp_replace(col, "[^0-9]", "") |
Limpa qualquer símbolo que não seja dígito numérico de 0 a 9. |
"111.222.333-44" ➔ "11122233344" |
F.when(cond, sim).otherwise(nao) |
Condicional distribuída (IF/ELSE) para tratar falhas de preenchimento. |
"N/A" ➔ 0.0 |
df.dropDuplicates(["cpf"]) |
Elimina cadastros repetidos mantendo apenas 1 registro por chave única. |
12 linhas brutas ➔ 10 clientes únicos |
Adicione uma célula extra no final do seu notebook no Google Colab e utilize o DataFrame limpo df_silver para responder às duas solicitações abaixo:
-
Filtro de Clientes VIP por Cidade: Exiba na tela apenas os clientes da cidade de
"São Paulo" cujo valor_gasto seja maior que R$ 500,00.
Comando de referência: df_silver.filter((F.col("cidade") == "São Paulo") & (F.col("valor_gasto") > 500)).show()
-
Auditoria de Qualidade de Cadastro: Conte quantos clientes vieram sem e-mail da base bruta e receberam o preenchimento automático
"sem_email@cadastro.com".
Comando de referência: df_silver.filter(F.col("email") == "sem_email@cadastro.com").count()