Curso Prático: Engenharia de Dados com Python e Google Cloud

Platform
Estrutura Pedagógica
Este curso segue uma progressão harmônica onde cada módulo representa 1.618x a complexidade do anterior,
respeitando a proporção áurea do aprendizado técnico.

Módulo 0: Preparação do Ambiente (1 semana)

Objetivos
Configurar ambiente local de desenvolvimento
Estabelecer conta GCP com free tier

Familiarizar-se com ferramentas essenciais

Ferramentas

bash

# Instalações necessárias
python 3.11+
pip install virtualenv
gcloud SDK
docker desktop
git
VSCode ou PyCharm
PostgreSQL local
MongoDB Community Edition

Projeto 0: Health Check

python
 # [Link]
import sys
import subprocess
import pkg_resources

def check_environment():
"""Verifica se o ambiente está configurado corretamente"""

checks = {
'Python': [Link],
'pip': [Link]('pip --version'),
'gcloud': [Link]('gcloud --version').split('\n')[0],
'docker': [Link]('docker --version'),
'postgres': [Link]('psql --version'),
'mongo': [Link]('mongod --version').split('\n')[0]
}

for tool, version in [Link]():

print(f"✓ {tool}: {version}")

if __name__ == "__main__":
check_environment()

Módulo 1: Python Essencial para Dados (2 semanas)

Conceitos Fundamentais
Estruturas de dados otimizadas
Manipulação eficiente com pandas

Conexões com bancos de dados

APIs REST

Projeto 1: Coletor de Dados Meteorológicos

python
# weather_collector.py
import requests
import pandas as pd
from datetime import datetime
import sqlite3
from typing import Dict, List

class WeatherCollector:
"""Coleta dados meteorológicos e armazena localmente"""

def __init__(self, db_path: str = "[Link]"):

self.db_path = db_path
self.api_base = "[Link]
self._init_database()

def _init_database(self):
"""Inicializa banco SQLite com schema otimizado"""
conn = [Link](self.db_path)
[Link]("""
CREATE TABLE IF NOT EXISTS weather_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
city TEXT NOT NULL,
timestamp INTEGER NOT NULL,
temperature REAL,
humidity INTEGER,
pressure INTEGER,
description TEXT,
UNIQUE(city, timestamp)
)
""")
[Link]("CREATE INDEX IF NOT EXISTS idx_city_time ON weather_data(city, timestamp)")
[Link]()
[Link]()

def collect_weather(self, cities: List[str], api_key: str) -> [Link]:

"""Coleta dados de múltiplas cidades"""
data = []

for city in cities:

response = [Link](
f"{self.api_base}/weather",
params={"q": city, "appid": api_key, "units": "metric"}
)

if response.status_code == 200:
weather = [Link]()
 [Link]({
'city': city,
'timestamp': int([Link]().timestamp()),
'temperature': weather['main']['temp'],
'humidity': weather['main']['humidity'],
'pressure': weather['main']['pressure'],
'description': weather['weather'][0]['description']
})

return [Link](data)

def save_to_database(self, df: [Link]):

"""Persiste dados com tratamento de duplicatas"""
conn = [Link](self.db_path)
df.to_sql('weather_data', conn, if_exists='append', index=False)
[Link]()

def analyze_trends(self, city: str, days: int = 7) -> Dict:

"""Análise básica de tendências"""
conn = [Link](self.db_path)

query = """
SELECT
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp,
AVG(humidity) as avg_humidity
FROM weather_data
WHERE city = ?
AND timestamp > strftime('%s', 'now', '-{} days')
""".format(days)

df = pd.read_sql_query(query, conn, params=[city])

[Link]()

return df.to_dict('records')[0]

# Uso prático
if __name__ == "__main__":
collector = WeatherCollector()

# Coletar dados
cities = ["São Paulo", "Rio de Janeiro", "Brasília"]
df = collector.collect_weather(cities, "YOUR_API_KEY")

# Persistir
collector.save_to_database(df)
 # Analisar
trends = collector.analyze_trends("São Paulo")
print(f"Tendências São Paulo: {trends}")

Módulo 2: SQL Avançado e Otimização (2 semanas)

Conceitos
Design de schemas eficientes

Índices e performance

Window functions

CTEs e queries complexas

Projeto 2: Sistema de Análise de Vendas

sql
-- [Link]
CREATE SCHEMA IF NOT EXISTS sales_analytics;

-- Tabela de produtos com particionamento

CREATE TABLE sales_analytics.products (
product_id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
category VARCHAR(100) NOT NULL,
price DECIMAL(10,2) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Tabela de vendas com índices otimizados

CREATE TABLE sales_analytics.sales (
sale_id BIGSERIAL PRIMARY KEY,
product_id INTEGER REFERENCES sales_analytics.products(product_id),
quantity INTEGER NOT NULL,
sale_date DATE NOT NULL,
customer_id INTEGER NOT NULL,
revenue DECIMAL(10,2) GENERATED ALWAYS AS (quantity * price) STORED
) PARTITION BY RANGE (sale_date);

-- Criar partições mensais

CREATE TABLE sales_analytics.sales_2024_01 PARTITION OF sales_analytics.sales
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

-- Índices compostos para queries frequentes

CREATE INDEX idx_sales_date_product ON sales_analytics.sales(sale_date, product_id);
CREATE INDEX idx_sales_customer_date ON sales_analytics.sales(customer_id, sale_date);

-- View materializada para dashboard

CREATE MATERIALIZED VIEW sales_analytics.daily_summary AS
WITH daily_stats AS (
SELECT
sale_date,
COUNT(DISTINCT customer_id) as unique_customers,
COUNT(*) as total_transactions,
SUM(quantity) as units_sold,
SUM(revenue) as total_revenue
FROM sales_analytics.sales
GROUP BY sale_date
),
moving_averages AS (
SELECT
sale_date,
unique_customers,
 total_transactions,
units_sold,
total_revenue,
AVG(total_revenue) OVER (
ORDER BY sale_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
) as revenue_7day_avg,
LAG(total_revenue, 7) OVER (ORDER BY sale_date) as revenue_week_ago
FROM daily_stats
)
SELECT
*,
CASE
WHEN revenue_week_ago > 0
THEN ((total_revenue - revenue_week_ago) / revenue_week_ago * 100)
ELSE 0
END as week_over_week_growth
FROM moving_averages;

-- Função para análise de cohort

CREATE OR REPLACE FUNCTION sales_analytics.cohort_analysis(
start_date DATE,
end_date DATE
) RETURNS TABLE (
cohort_month DATE,
months_since_first_purchase INTEGER,
customers INTEGER,
revenue DECIMAL
) AS $$
BEGIN
RETURN QUERY
WITH first_purchase AS (
SELECT
customer_id,
DATE_TRUNC('month', MIN(sale_date))::DATE as cohort_month
FROM sales_analytics.sales
WHERE sale_date BETWEEN start_date AND end_date
GROUP BY customer_id
),
cohort_data AS (
SELECT
fp.cohort_month,
DATE_PART('month', AGE(DATE_TRUNC('month', s.sale_date), fp.cohort_month))::INTEGER as months_since,
COUNT(DISTINCT s.customer_id) as customers,
SUM([Link]) as revenue
FROM first_purchase fp
JOIN sales_analytics.sales s ON fp.customer_id = s.customer_id
 GROUP BY fp.cohort_month, months_since
)
SELECT * FROM cohort_data
ORDER BY cohort_month, months_since_first_purchase;
END;
$$ LANGUAGE plpgsql;

python
# sales_analyzer.py
import psycopg2
import pandas as pd
from contextlib import contextmanager
from typing import Generator

class SalesAnalyzer:
"""Analisador de vendas com SQL otimizado"""

def __init__(self, connection_string: str):

self.connection_string = connection_string

@contextmanager
def get_connection(self) -> Generator:
"""Context manager para conexões seguras"""
conn = [Link](self.connection_string)
try:
yield conn
finally:
[Link]()

def load_sales_data(self, df: [Link]):

"""Carrega dados em lote com COPY"""
with self.get_connection() as conn:
cursor = [Link]()

# Criar tabela temporária

[Link]("""
CREATE TEMP TABLE temp_sales (
product_id INTEGER,
quantity INTEGER,
sale_date DATE,
customer_id INTEGER
)
""")

# COPY em lote (muito mais rápido que INSERT)

from io import StringIO
buffer = StringIO()
df.to_csv(buffer, index=False, header=False)
[Link](0)

cursor.copy_from(
buffer,
'temp_sales',
sep=',',
 columns=['product_id', 'quantity', 'sale_date', 'customer_id']
)

# Inserir com tratamento de conflitos

[Link]("""
INSERT INTO sales_analytics.sales
(product_id, quantity, sale_date, customer_id)
SELECT * FROM temp_sales
ON CONFLICT DO NOTHING
""")

[Link]()

def get_product_performance(self, start_date: str, end_date: str) -> [Link]:

"""Análise de performance por produto usando window functions"""
query = """
WITH product_metrics AS (
SELECT
[Link],
[Link],
SUM([Link]) as total_quantity,
SUM([Link]) as total_revenue,
COUNT(DISTINCT s.customer_id) as unique_customers,
AVG([Link] / [Link]) as avg_price
FROM sales_analytics.sales s
JOIN sales_analytics.products p ON s.product_id = p.product_id
WHERE s.sale_date BETWEEN %s AND %s
GROUP BY [Link], [Link]
),
ranked_products AS (
SELECT
*,
RANK() OVER (PARTITION BY category ORDER BY total_revenue DESC) as revenue_rank,
PERCENT_RANK() OVER (ORDER BY total_revenue) as revenue_percentile
FROM product_metrics
)
SELECT * FROM ranked_products
ORDER BY category, revenue_rank
"""

with self.get_connection() as conn:

return pd.read_sql_query(query, conn, params=[start_date, end_date])
Módulo 3: NoSQL e Modelagem de Documentos (2 semanas)

Conceitos
Modelagem orientada a queries

Sharding e replicação

Índices em NoSQL

Patterns de agregação

Projeto 3: Sistema de Recomendação com MongoDB

python
# recommendation_engine.py
from pymongo import MongoClient, ASCENDING, TEXT
from datetime import datetime, timedelta
import numpy as np
from typing import List, Dict, Optional
from bson import ObjectId

class RecommendationEngine:
"""Motor de recomendação usando MongoDB"""

def __init__(self, connection_string: str):

[Link] = MongoClient(connection_string)
[Link] = [Link].recommendation_db
self._setup_collections()

def _setup_collections(self):
"""Configura coleções com índices otimizados"""

# Coleção de usuários
if 'users' not in [Link].list_collection_names():
[Link].create_collection('users')
[Link].create_index([('email', ASCENDING)], unique=True)
[Link].create_index([('created_at', ASCENDING)])

# Coleção de produtos com índice de texto

if 'products' not in [Link].list_collection_names():
[Link].create_collection('products')
[Link].create_index([('name', TEXT), ('description', TEXT)])
[Link].create_index([('category', ASCENDING)])
[Link].create_index([('tags', ASCENDING)])

# Coleção de interações
if 'interactions' not in [Link].list_collection_names():
[Link].create_collection('interactions')
[Link].create_index([
('user_id', ASCENDING),
('timestamp', ASCENDING)
])
[Link].create_index([
('product_id', ASCENDING),
('type', ASCENDING)
])
# Índice composto para queries de agregação
[Link].create_index([
('user_id', ASCENDING),
('product_id', ASCENDING),
 ('type', ASCENDING)
])

def track_interaction(self, user_id: str, product_id: str,

interaction_type: str, metadata: Dict = None):
"""Registra interação com deduplicação"""

interaction = {
'user_id': ObjectId(user_id),
'product_id': ObjectId(product_id),
'type': interaction_type, # view, click, purchase, rating
'timestamp': [Link](),
'metadata': metadata or {}
}

# Adicionar score baseado no tipo de interação

scores = {'view': 1, 'click': 2, 'purchase': 5, 'rating': 3}
interaction['score'] = [Link](interaction_type, 1)

# Upsert para evitar duplicatas em janela de tempo

[Link].update_one(
{
'user_id': interaction['user_id'],
'product_id': interaction['product_id'],
'timestamp': {
'$gte': [Link]() - timedelta(hours=1)
}
},
{'$set': interaction},
upsert=True
)

def get_collaborative_recommendations(self, user_id: str, limit: int = 10) -> List[Dict]:

"""Recomendações colaborativas usando agregação"""

pipeline = [
# Encontrar produtos que o usuário interagiu
{
'$match': {
'user_id': ObjectId(user_id)
}
},
# Agrupar por produto
{
'$group': {
'_id': '$product_id',
'user_score': {'$sum': '$score'}
 }
},
# Encontrar outros usuários que interagiram com os mesmos produtos
{
'$lookup': {
'from': 'interactions',
'localField': '_id',
'foreignField': 'product_id',
'as': 'other_users'
}
},
# Expandir array de outros usuários
{
'$unwind': '$other_users'
},
# Filtrar o próprio usuário
{
'$match': {
'other_users.user_id': {'$ne': ObjectId(user_id)}
}
},
# Encontrar produtos que esses usuários também gostaram
{
'$lookup': {
'from': 'interactions',
'let': {'other_user': '$other_users.user_id'},
'pipeline': [
{
'$match': {
'$expr': {
'$and': [
{'$eq': ['$user_id', '$$other_user']},
{'$gte': ['$score', 3]}
]
}
}
}
],
'as': 'recommendations'
}
},
# Expandir recomendações
{
'$unwind': '$recommendations'
},
# Agrupar e calcular score de recomendação
{
 '$group': {
'_id': '$recommendations.product_id',
'recommendation_score': {
'$sum': {
'$multiply': [
'$user_score',
'$[Link]',
'$other_users.score'
]
}
},
'support': {'$sum': 1} # Número de usuários que recomendam
}
},
# Filtrar produtos já vistos
{
'$lookup': {
'from': 'interactions',
'let': {'prod_id': '$_id'},
'pipeline': [
{
'$match': {
'$expr': {
'$and': [
{'$eq': ['$product_id', '$$prod_id']},
{'$eq': ['$user_id', ObjectId(user_id)]}
]
}
}
}
],
'as': 'already_seen'
}
},
{
'$match': {
'already_seen': {'$size': 0}
}
},
# Buscar detalhes do produto
{
'$lookup': {
'from': 'products',
'localField': '_id',
'foreignField': '_id',
'as': 'product'
}
 },
{
'$unwind': '$product'
},
# Ordenar por score e limitar
{
'$sort': {
'recommendation_score': -1,
'support': -1
}
},
{
'$limit': limit
},
# Formatar saída
{
'$project': {
'product_id': '$_id',
'name': '$[Link]',
'category': '$[Link]',
'score': '$recommendation_score',
'recommended_by': '$support'
}
}
]

return list([Link](pipeline))

def get_content_based_recommendations(self, product_id: str, limit: int = 10) -> List[Dict]:

"""Recomendações baseadas em conteúdo usando índices de texto"""

# Buscar produto original

product = [Link].find_one({'_id': ObjectId(product_id)})
if not product:
return []

# Busca por similaridade usando texto e tags

pipeline = [
{
'$match': {
'$and': [
{'_id': {'$ne': ObjectId(product_id)}},
{
'$or': [
{'$text': {'$search': [Link]('name', '')}},
{'category': [Link]('category')},
{'tags': {'$in': [Link]('tags', [])}}
 ]
}
]
}
},
# Calcular score de similaridade
{
'$addFields': {
'similarity_score': {
'$add': [
{'$cond': [{'$eq': ['$category', [Link]('category')]}, 10, 0]},
{
'$size': {
'$setIntersection': [
'$tags',
[Link]('tags', [])
]
}
}
]
}
}
},
# Adicionar popularidade
{
'$lookup': {
'from': 'interactions',
'localField': '_id',
'foreignField': 'product_id',
'as': 'interactions'
}
},
{
'$addFields': {
'popularity': {'$size': '$interactions'}
}
},
# Score final combinando similaridade e popularidade
{
'$addFields': {
'final_score': {
'$add': [
{'$multiply': ['$similarity_score', 0.7]},
{'$multiply': ['$popularity', 0.3]}
]
}
}
 },
{
'$sort': {'final_score': -1}
},
{
'$limit': limit
},
{
'$project': {
'product_id': '$_id',
'name': 1,
'category': 1,
'tags': 1,
'score': '$final_score'
}
}
]

return list([Link](pipeline))

Módulo 4: Google Cloud Platform - Fundamentos (3 semanas)

Conceitos
Arquitetura serverless

Processamento em lote vs streaming

Segurança e IAM

Monitoramento e logging

Projeto 4.1: Pipeline de Dados Serverless

python
# cloud_functions/data_ingestion/[Link]
import functions_framework
from [Link] import storage, bigquery, firestore
import pandas as pd
import json
from datetime import datetime
from typing import Dict, Any

# Inicializar clientes
storage_client = [Link]()
bigquery_client = [Link]()
firestore_client = [Link]()

@functions_framework.cloud_event
def process_uploaded_file(cloud_event):
"""
Cloud Function triggerada por upload no Cloud Storage
Processa arquivo e carrega no BigQuery
"""

# Extrair informações do evento

data = cloud_event.data
bucket_name = data['bucket']
file_name = data['name']

# Validar tipo de arquivo

if not file_name.endswith(('.csv', '.json')):
print(f"Arquivo ignorado: {file_name}")
return

try:
# Baixar arquivo do Storage
bucket = storage_client.bucket(bucket_name)
blob = [Link](file_name)

# Processar baseado no tipo

if file_name.endswith('.csv'):
df = pd.read_csv([Link]('r'))
else:
df = pd.read_json([Link]('r'))

# Adicionar metadados
df['processed_at'] = [Link]()
df['source_file'] = file_name

# Validação básica
 df = validate_and_clean_data(df)

# Carregar no BigQuery
dataset_id = 'raw_data'
table_id = extract_table_name(file_name)

table_ref = f"{bigquery_client.project}.{dataset_id}.{table_id}"

job_config = [Link](
write_disposition=[Link].WRITE_APPEND,
schema_update_options=[
[Link].ALLOW_FIELD_ADDITION
]
)

job = bigquery_client.load_table_from_dataframe(
df, table_ref, job_config=job_config
)
[Link]() # Aguardar conclusão

# Registrar no Firestore
doc_ref = firestore_client.collection('processing_log').document()
doc_ref.set({
'file_name': file_name,
'records_processed': len(df),
'status': 'success',
'processed_at': [Link](),
'bigquery_table': table_ref
})

print(f"Processado com sucesso: {file_name} ({len(df)} registros)")

except Exception as e:
# Log de erro no Firestore
error_ref = firestore_client.collection('processing_errors').document()
error_ref.set({
'file_name': file_name,
'error': str(e),
'timestamp': [Link]()
})

raise e

def validate_and_clean_data(df: [Link]) -> [Link]:

"""Validação e limpeza básica de dados"""

# Remover duplicatas completas

 df = df.drop_duplicates()

# Converter tipos de dados

for col in df.select_dtypes(include=['object']).columns:
# Tentar converter para datetime
try:
df[col] = pd.to_datetime(df[col])
except:
pass

# Remover espaços em branco extras

for col in df.select_dtypes(include=['object']).columns:
df[col] = df[col].[Link]()

return df

def extract_table_name(file_name: str) -> str:

"""Extrai nome da tabela do nome do arquivo"""
# Exemplo: sales_2024_01.csv -> sales
parts = file_name.split('_')
return parts[0] if parts else 'unknown'

# [Link]
"""
functions-framework==3.*
google-cloud-storage==2.10.*
google-cloud-bigquery==3.11.*
google-cloud-firestore==2.11.*
pandas==2.0.*
pyarrow==12.0.*
"""

Projeto 4.2: API de Analytics com App Engine

python
# app_engine/[Link]
from flask import Flask, request, jsonify
from [Link] import bigquery
from [Link] import firestore
import pandas as pd
from datetime import datetime, timedelta
import numpy as np
from functools import lru_cache
import json

app = Flask(__name__)
bigquery_client = [Link]()
firestore_client = [Link]()

# Cache para queries frequentes

@lru_cache(maxsize=128)
def get_cached_query_result(query_hash: str, cache_duration: int = 3600):
"""Cache de resultados de queries no Firestore"""

cache_ref = firestore_client.collection('query_cache').document(query_hash)
cache_doc = cache_ref.get()

if cache_doc.exists:
cache_data = cache_doc.to_dict()
cache_time = cache_data.get('timestamp', [Link])

if [Link]() - cache_time < timedelta(seconds=cache_duration):

return cache_data.get('result')

return None

@[Link]('/api/v1/analytics/sales', methods=['GET'])
def sales_analytics():
"""Endpoint para análise de vendas"""

# Parâmetros da query
start_date = [Link]('start_date',
([Link]() - timedelta(days=30)).strftime('%Y-%m-%d'))
end_date = [Link]('end_date',
[Link]().strftime('%Y-%m-%d'))
granularity = [Link]('granularity', 'daily') # daily, weekly, monthly
metrics = [Link]('metrics') or ['revenue', 'orders', 'avg_order_value']

# Hash para cache

query_hash = f"sales_{start_date}_{end_date}_{granularity}_{'_'.join(metrics)}"
# Verificar cache
cached_result = get_cached_query_result(query_hash)
if cached_result:
return jsonify(cached_result)

# Query BigQuery
query = f"""
WITH sales_data AS (
SELECT
DATE(sale_timestamp) as sale_date,
order_id,
customer_id,
total_amount,
items_count
FROM `{bigquery_client.project}.[Link]`
WHERE DATE(sale_timestamp) BETWEEN @start_date AND @end_date
),
aggregated AS (
SELECT
{get_date_trunc(granularity)} as period,
COUNT(DISTINCT order_id) as orders,
COUNT(DISTINCT customer_id) as unique_customers,
SUM(total_amount) as revenue,
AVG(total_amount) as avg_order_value,
SUM(items_count) as total_items
FROM sales_data
GROUP BY period
)
SELECT
period,
{', '.join([f'{m} as {m}' for m in metrics if m in
['orders', 'unique_customers', 'revenue', 'avg_order_value', 'total_items']])}
FROM aggregated
ORDER BY period
"""

job_config = [Link](
query_parameters=[
[Link]("start_date", "DATE", start_date),
[Link]("end_date", "DATE", end_date)
]
)

query_job = bigquery_client.query(query, job_config=job_config)

results = query_job.result()

# Converter para formato JSON

 data = []
for row in results:
[Link]({
'period': [Link]() if hasattr([Link], 'isoformat') else str([Link]),
**{metric: float(getattr(row, metric, 0)) for metric in metrics}
})

response = {
'data': data,
'metadata': {
'start_date': start_date,
'end_date': end_date,
'granularity': granularity,
'metrics': metrics,
'generated_at': [Link]().isoformat()
}
}

# Salvar no cache
cache_ref = firestore_client.collection('query_cache').document(query_hash)
cache_ref.set({
'result': response,
'timestamp': [Link]()
})

return jsonify(response)

@[Link]('/api/v1/analytics/forecast', methods=['POST'])
def sales_forecast():
"""Previsão de vendas usando histórico"""

data = request.get_json()
product_id = [Link]('product_id')
days_ahead = [Link]('days_ahead', 7)

# Query histórico
query = """
SELECT
DATE(sale_timestamp) as date,
SUM(quantity) as daily_sales
FROM `{}.analytics.sales_items`
WHERE product_id = @product_id
AND DATE(sale_timestamp) >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY date
ORDER BY date
""".format(bigquery_client.project)
job_config = [Link](
query_parameters=[
[Link]("product_id", "STRING", product_id)
]
)

df = bigquery_client.query(query, job_config=job_config).to_dataframe()

if len(df) < 30:

return jsonify({'error': 'Dados insuficientes para previsão'}), 400

# Modelo simples de média móvel ponderada

df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)

# Calcular tendência
df['rolling_mean_7'] = df['daily_sales'].rolling(window=7).mean()
df['rolling_mean_30'] = df['daily_sales'].rolling(window=30).mean()

# Peso maior para dados recentes

recent_avg = df['daily_sales'].tail(7).mean()
medium_avg = df['daily_sales'].tail(30).mean()
long_avg = df['daily_sales'].mean()

# Previsão ponderada
forecast = (recent_avg * 0.5 + medium_avg * 0.3 + long_avg * 0.2)

# Gerar previsões
future_dates = pd.date_range(
start=[Link]() + timedelta(days=1),
periods=days_ahead,
freq='D'
)

predictions = []
for date in future_dates:
# Adicionar sazonalidade semanal
day_of_week = [Link]
seasonality_factor = 1.0

# Fins de semana geralmente têm vendas diferentes

if day_of_week in [5, 6]: # Sábado e Domingo
seasonality_factor = 0.8

[Link]({
'date': [Link]('%Y-%m-%d'),
'predicted_sales': round(forecast * seasonality_factor, 2),
 'confidence_interval': {
'lower': round(forecast * seasonality_factor * 0.8, 2),
'upper': round(forecast * seasonality_factor * 1.2, 2)
}
})

return jsonify({
'product_id': product_id,
'forecast': predictions,
'model_metrics': {
'historical_average': round(df['daily_sales'].mean(), 2),
'recent_trend': 'increasing' if recent_avg > long_avg else 'decreasing',
'volatility': round(df['daily_sales'].std(), 2)
}
})

def get_date_trunc(granularity: str) -> str:

"""Retorna função SQL para truncar data"""
mapping = {
'daily': 'sale_date',
'weekly': 'DATE_TRUNC(sale_date, WEEK)',
'monthly': 'DATE_TRUNC(sale_date, MONTH)'
}
return [Link](granularity, 'sale_date')

@[Link](Exception)
def handle_error(error):
"""Handler global de erros"""

# Log no Firestore
error_ref = firestore_client.collection('api_errors').document()
error_ref.set({
'endpoint': [Link],
'method': [Link],
'error': str(error),
'timestamp': [Link]()
})

return jsonify({
'error': 'Internal server error',
'message': str(error) if [Link] else 'An error occurred'
}), 500

if __name__ == '__main__':
[Link](host='[Link]', port=8080)

# [Link]
 """
runtime: python311
instance_class: F2

automatic_scaling:
target_cpu_utilization: 0.65
min_instances: 1
max_instances: 10

env_variables:
GAE_USE_SOCKETS_FOR_CLOUDSQL: 'true'

handlers:
- url: /api/.*
script: auto
secure: always
"""

Projeto 4.3: Pipeline de ML com Vertex AI

python
# vertex_ai_pipeline/[Link]
from [Link] import aiplatform
from [Link] import bigquery
from [Link] import storage
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from [Link] import StandardScaler
from [Link] import RandomForestRegressor
import joblib
from datetime import datetime
import json

class MLPipeline:
"""Pipeline de ML integrado com GCP"""

def __init__(self, project_id: str, location: str = 'us-central1'):

self.project_id = project_id
[Link] = location

[Link](project=project_id, location=location)
self.bq_client = [Link](project=project_id)
self.storage_client = [Link](project=project_id)

def prepare_training_data(self, dataset_id: str, table_id: str) -> [Link]:

"""Prepara dados do BigQuery para treinamento"""

query = f"""
WITH features AS (
SELECT
customer_id,
COUNT(DISTINCT order_id) as total_orders,
SUM(total_amount) as lifetime_value,
AVG(total_amount) as avg_order_value,
MAX(order_date) as last_order_date,
MIN(order_date) as first_order_date,
COUNT(DISTINCT product_category) as categories_purchased,
COUNT(DISTINCT DATE_TRUNC(order_date, MONTH)) as active_months
FROM `{self.project_id}.{dataset_id}.{table_id}`
GROUP BY customer_id
),
target AS (
SELECT
customer_id,
SUM(total_amount) as next_month_value
FROM `{self.project_id}.{dataset_id}.{table_id}`
 WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY customer_id
)
SELECT
f.*,
DATE_DIFF(CURRENT_DATE(), f.last_order_date, DAY) as days_since_last_order,
DATE_DIFF(f.last_order_date, f.first_order_date, DAY) as customer_lifetime_days,
COALESCE(t.next_month_value, 0) as target
FROM features f
LEFT JOIN target t ON f.customer_id = t.customer_id
"""

df = self.bq_client.query(query).to_dataframe()

# Feature engineering adicional

df['orders_per_month'] = df['total_orders'] / (df['active_months'] + 1)
df['avg_days_between_orders'] = df['customer_lifetime_days'] / (df['total_orders'] + 1)

return df

def train_model(self, df: [Link], model_name: str):

"""Treina modelo e registra no Vertex AI"""

# Preparar features e target

feature_columns = [col for col in [Link]
if col not in ['customer_id', 'target', 'first_order_date', 'last_order_date']]

X = df[feature_columns].fillna(0)
y = df['target']

# Split dados
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)

# Normalizar features
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = [Link](X_test)

# Treinar modelo
model = RandomForestRegressor(
n_estimators=100,
max_depth=10,
random_state=42,
n_jobs=-1
)
[Link](X_train_scaled, y_train)

# Avaliar modelo
train_score = [Link](X_train_scaled, y_train)
test_score = [Link](X_test_scaled, y_test)

print(f"Train R²: {train_score:.4f}")

print(f"Test R²: {test_score:.4f}")

# Salvar artefatos
timestamp = [Link]().strftime('%Y%m%d_%H%M%S')
model_path = f"models/{model_name}_{timestamp}"

# Salvar no Cloud Storage

bucket_name = f"{self.project_id}-ml-models"
bucket = self.storage_client.bucket(bucket_name)

# Salvar modelo
model_blob = [Link](f"{model_path}/[Link]")
model_blob.upload_from_string([Link](model))

# Salvar scaler
scaler_blob = [Link](f"{model_path}/[Link]")
scaler_blob.upload_from_string([Link](scaler))

# Salvar metadados
metadata = {
'model_name': model_name,
'timestamp': timestamp,
'features': feature_columns,
'train_score': train_score,
'test_score': test_score,
'train_samples': len(X_train),
'test_samples': len(X_test)
}

metadata_blob = [Link](f"{model_path}/[Link]")
metadata_blob.upload_from_string([Link](metadata))

# Registrar no Model Registry do Vertex AI

model_display_name = f"{model_name}-{timestamp}"

model = [Link](
display_name=model_display_name,
artifact_uri=f"gs://{bucket_name}/{model_path}",
serving_container_image_uri="[Link]/vertex-ai/prediction/sklearn-cpu.1-0:latest"
)
 return model

def deploy_model(self, model: [Link], endpoint_name: str):

"""Deploy modelo para endpoint de predição"""

# Criar ou reutilizar endpoint

endpoints = [Link](
filter=f'display_name="{endpoint_name}"'
)

if endpoints:
endpoint = endpoints[0]
else:
endpoint = [Link](
display_name=endpoint_name,
description="Endpoint para previsão de valor do cliente"
)

# Deploy do modelo
deployed_model = [Link](
model=model,
deployed_model_display_name=model.display_name,
machine_type="n1-standard-2",
min_replica_count=1,
max_replica_count=3,
accelerator_type=None,
accelerator_count=0
)

return endpoint

def batch_predict(self, model: [Link],

input_dataset: str, output_dataset: str):
"""Predição em lote usando BigQuery"""

job = model.batch_predict(
job_display_name=f"batch-prediction-{[Link]().strftime('%Y%m%d-%H%M%S')}",
bigquery_source=input_dataset,
bigquery_destination_prefix=output_dataset,
machine_type="n1-standard-4",
starting_replica_count=1,
max_replica_count=5
)

[Link]()
 return job

# Uso do pipeline
if __name__ == "__main__":
pipeline = MLPipeline(project_id="seu-projeto-gcp")

# Preparar dados
df = pipeline.prepare_training_data("ecommerce", "orders")

# Treinar modelo
model = pipeline.train_model(df, "customer_ltv_predictor")

# Deploy
endpoint = pipeline.deploy_model(model, "customer-ltv-endpoint")

print(f"Modelo deployado em: {endpoint.resource_name}")

Módulo 5: Integração Completa (3 semanas)

Projeto Final: Plataforma de Analytics em Tempo Real

Este projeto integra todos os conceitos aprendidos em uma solução completa.

python
# architecture/data_platform.py
"""
Arquitetura da Plataforma de Dados

Componentes:
1. Ingestão: Cloud Functions + Pub/Sub
2. Processamento: Dataflow + BigQuery
3. Armazenamento: Cloud SQL + Firestore + BigQuery
4. Análise: Vertex AI + BigQuery ML
5. Visualização: Looker Studio + API customizada
6. Monitoramento: Cloud Monitoring + Logging
"""

from [Link] import pubsub_v1, bigquery, firestore

from [Link] import dataflow_v1beta3 as dataflow
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.[Link] import WriteToBigQuery
import json
from datetime import datetime
from typing import Dict, Any, Iterator

class DataPlatform:
"""Plataforma integrada de dados em GCP"""

def __init__(self, project_id: str):

self.project_id = project_id
[Link] = pubsub_v1.PublisherClient()
self.bq_client = [Link]()
self.fs_client = [Link]()

def create_infrastructure(self):
"""Cria toda a infraestrutura necessária"""

# Criar tópicos Pub/Sub

topics = ['raw-events', 'processed-events', 'alerts']
for topic_name in topics:
topic_path = [Link].topic_path(self.project_id, topic_name)
try:
[Link].create_topic(request={"name": topic_path})
print(f"Tópico criado: {topic_name}")
except Exception as e:
print(f"Tópico já existe: {topic_name}")

# Criar datasets BigQuery

datasets = [
 ('raw_data', 'Dados brutos de eventos'),
('processed_data', 'Dados processados e agregados'),
('ml_features', 'Features para modelos de ML')
]

for dataset_id, description in datasets:

dataset = [Link](f"{self.project_id}.{dataset_id}")
[Link] = description
[Link] = "US"

try:
dataset = self.bq_client.create_dataset(dataset, timeout=30)
print(f"Dataset criado: {dataset_id}")
except Exception:
print(f"Dataset já existe: {dataset_id}")

# Criar coleções Firestore

collections = ['events_metadata', 'processing_status', 'model_registry']
for collection in collections:
# Criar documento dummy para inicializar coleção
doc_ref = self.fs_client.collection(collection).document('_init')
doc_ref.set({'created_at': [Link]()})
print(f"Coleção criada: {collection}")

# Dataflow Pipeline
class EventProcessor([Link]):
"""Processa eventos em streaming"""

def process(self, element: bytes) -> Iterator[Dict[str, Any]]:

try:
# Decodificar mensagem
event = [Link]([Link]('utf-8'))

# Enriquecer evento
event['processed_at'] = [Link]().isoformat()
event['processing_version'] = '1.0'

# Validar campos obrigatórios

required_fields = ['event_id', 'user_id', 'event_type', 'timestamp']
if all(field in event for field in required_fields):
yield event
else:
# Enviar para dead letter queue
yield [Link]('invalid_events', event)

except Exception as e:
# Log de erro
 error_event = {
'error': str(e),
'raw_data': [Link]('utf-8', errors='ignore'),
'timestamp': [Link]().isoformat()
}
yield [Link]('errors', error_event)

class AggregateEvents([Link]):
"""Agrega eventos por janela de tempo"""

def process(self, element: tuple) -> Iterator[Dict[str, Any]]:

key, events = element
events_list = list(events)

aggregation = {
'window_start': [Link].to_utc_datetime().isoformat(),
'window_end': [Link].to_utc_datetime().isoformat(),
'event_count': len(events_list),
'unique_users': len(set(e['user_id'] for e in events_list)),
'event_types': {}
}

# Contar por tipo de evento

for event in events_list:
event_type = [Link]('event_type', 'unknown')
aggregation['event_types'][event_type] = \
aggregation['event_types'].get(event_type, 0) + 1

yield aggregation

def create_streaming_pipeline(project_id: str, subscription: str):

"""Cria pipeline de streaming com Dataflow"""

pipeline_options = PipelineOptions(
project=project_id,
runner='DataflowRunner',
temp_location=f'gs://{project_id}-dataflow-temp/temp',
region='us-central1',
streaming=True,
save_main_session=True
)

# Schema BigQuery
event_schema = {
'fields': [
{'name': 'event_id', 'type': 'STRING', 'mode': 'REQUIRED'},
{'name': 'user_id', 'type': 'STRING', 'mode': 'REQUIRED'},
 {'name': 'event_type', 'type': 'STRING', 'mode': 'REQUIRED'},
{'name': 'timestamp', 'type': 'TIMESTAMP', 'mode': 'REQUIRED'},
{'name': 'properties', 'type': 'JSON', 'mode': 'NULLABLE'},
{'name': 'processed_at', 'type': 'TIMESTAMP', 'mode': 'REQUIRED'}
]
}

aggregation_schema = {
'fields': [
{'name': 'window_start', 'type': 'TIMESTAMP', 'mode': 'REQUIRED'},
{'name': 'window_end', 'type': 'TIMESTAMP', 'mode': 'REQUIRED'},
{'name': 'event_count', 'type': 'INTEGER', 'mode': 'REQUIRED'},
{'name': 'unique_users', 'type': 'INTEGER', 'mode': 'REQUIRED'},
{'name': 'event_types', 'type': 'JSON', 'mode': 'REQUIRED'}
]
}

with [Link](options=pipeline_options) as pipeline:

# Ler do Pub/Sub
events = (
pipeline
| 'Read from PubSub' >> [Link](
subscription=f'projects/{project_id}/subscriptions/{subscription}'
)
| 'Process Events' >> [Link](EventProcessor()).with_outputs(
'invalid_events', 'errors', main='valid_events'
)
)

# Processar eventos válidos

valid_events = events.valid_events

# Escrever eventos processados no BigQuery

(
valid_events
| 'Write to BigQuery' >> WriteToBigQuery(
f'{project_id}:processed_data.events',
schema=event_schema,
write_disposition=[Link].WRITE_APPEND,
create_disposition=[Link].CREATE_IF_NEEDED
)
)

# Agregações por janela de 1 minuto

(
valid_events
| 'Add Timestamp' >> [Link](
 lambda x: [Link](
x,
[Link](x['timestamp']).timestamp()
)
)
| 'Window' >> [Link](
[Link](60) # Janela de 60 segundos
)
| 'Group by Window' >> [Link]()
| 'Aggregate' >> [Link](AggregateEvents())
| 'Write Aggregations' >> WriteToBigQuery(
f'{project_id}:processed_data.event_aggregations',
schema=aggregation_schema,
write_disposition=[Link].WRITE_APPEND,
create_disposition=[Link].CREATE_IF_NEEDED
)
)

# Tratar eventos inválidos

(
events.invalid_events
| 'Invalid to JSON' >> [Link]([Link])
| 'Write Invalid to Storage' >> [Link](
f'gs://{project_id}-invalid-events/invalid',
file_name_suffix='.json',
num_shards=1
)
)

Módulo 6: Otimização e Boas Práticas (2 semanas)

Conceitos Avançados
Query optimization
Cost management

Security best practices

CI/CD para dados

Projeto 6: Framework de Monitoramento e Otimização

python
# monitoring/platform_monitor.py
from [Link] import monitoring_v3
from [Link] import logging
from [Link] import bigquery
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import smtplib
from [Link] import MIMEText

class PlatformMonitor:
"""Sistema de monitoramento e otimização da plataforma"""

def __init__(self, project_id: str):

self.project_id = project_id
self.monitoring_client = monitoring_v3.MetricServiceClient()
self.logging_client = [Link]()
self.bq_client = [Link]()
self.project_path = f"projects/{project_id}"

def analyze_bigquery_costs(self, days: int = 30) -> [Link]:

"""Analisa custos do BigQuery"""

query = f"""
SELECT
user_email,
DATE(creation_time) as query_date,
query,
total_bytes_processed,
total_slot_ms,
ROUND(total_bytes_processed / POW(10, 12) * 5, 2) as estimated_cost_usd
FROM `{self.project_id}.region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT`
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {days} DAY)
AND state = 'DONE'
AND statement_type = 'SELECT'
ORDER BY total_bytes_processed DESC
"""

df = self.bq_client.query(query).to_dataframe()

# Identificar queries problemáticas

df['is_expensive'] = df['estimated_cost_usd'] > 1.0
df['bytes_per_slot'] = df['total_bytes_processed'] / (df['total_slot_ms'] + 1)

return df
def optimize_queries(self, expensive_queries: [Link]) -> List[Dict]:
"""Sugere otimizações para queries caras"""

optimizations = []

for _, query_info in expensive_queries.iterrows():

query_text = query_info['query'].lower()
suggestions = []

# Verificar SELECT *
if 'select *' in query_text:
[Link]({
'issue': 'SELECT * detectado',
'suggestion': 'Especifique apenas as colunas necessárias',
'potential_savings': '60-90%'
})

# Verificar falta de particionamento

if 'where' not in query_text or 'partition' not in query_text:
[Link]({
'issue': 'Query sem filtro de partição',
'suggestion': 'Adicione filtros de data para usar particionamento',
'potential_savings': '70-95%'
})

# Verificar JOINs desnecessários

join_count = query_text.count('join')
if join_count > 3:
[Link]({
'issue': f'{join_count} JOINs detectados',
'suggestion': 'Considere materializar views ou redesenhar schema',
'potential_savings': '30-50%'
})

if suggestions:
[Link]({
'query_date': query_info['query_date'],
'user': query_info['user_email'],
'current_cost': query_info['estimated_cost_usd'],
'suggestions': suggestions
})

return optimizations

def create_monitoring_dashboard(self):
"""Cria métricas customizadas e alertas"""
 # Métrica customizada para latência de pipeline
descriptor = monitoring_v3.MetricDescriptor(
type=f"[Link]/{self.project_id}/pipeline_latency",
metric_kind=monitoring_v3.[Link],
value_type=monitoring_v3.[Link],
description="Latência do pipeline de dados em segundos",
display_name="Pipeline Latency"
)

try:
self.monitoring_client.create_metric_descriptor(
name=self.project_path,
metric_descriptor=descriptor
)
except Exception:
pass # Métrica já existe

# Criar política de alerta

alert_policy = monitoring_v3.AlertPolicy(
display_name="High Pipeline Latency",
conditions=[
monitoring_v3.[Link](
display_name="Pipeline latency > 5 minutes",
condition_threshold=monitoring_v3.[Link](
filter=f'[Link]="[Link]/{self.project_id}/pipeline_latency"',
comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
threshold_value=300, # 5 minutos
duration=monitoring_v3.Duration(seconds=60),
aggregations=[
monitoring_v3.Aggregation(
alignment_period=monitoring_v3.Duration(seconds=60),
per_series_aligner=monitoring_v3.[Link].ALIGN_MEAN
)
]
)
)
]
)

return alert_policy

def performance_report(self) -> Dict:

"""Gera relatório completo de performance"""

report = {
'generated_at': [Link]().isoformat(),
'period': 'last_30_days',
 'metrics': {}
}

# BigQuery Performance
bq_query = """
SELECT
COUNT(*) as total_queries,
SUM(total_bytes_processed) / POW(10, 12) as tb_processed,
AVG(total_slot_ms) / 1000 as avg_slot_seconds,
SUM(total_bytes_processed) / POW(10, 12) * 5 as total_cost_usd
FROM `region-us.INFORMATION_SCHEMA.JOBS_BY_PROJECT`
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
AND state = 'DONE'
"""

bq_metrics = self.bq_client.query(bq_query).to_dataframe().iloc[0].to_dict()
report['metrics']['bigquery'] = bq_metrics

# Storage Usage
storage_query = """
SELECT
schema_name as dataset,
SUM(size_bytes) / POW(10, 9) as size_gb
FROM `region-us.INFORMATION_SCHEMA.TABLE_STORAGE`
GROUP BY dataset
ORDER BY size_gb DESC
"""

storage_df = self.bq_client.query(storage_query).to_dataframe()
report['metrics']['storage'] = {
'total_gb': storage_df['size_gb'].sum(),
'by_dataset': storage_df.to_dict('records')
}

# Pipeline Health
recent_errors = self._get_recent_errors()
report['metrics']['pipeline_health'] = {
'error_count': len(recent_errors),
'error_rate': len(recent_errors) / max(bq_metrics['total_queries'], 1),
'top_errors': recent_errors[:5]
}

return report

def _get_recent_errors(self, hours: int = 24) -> List[Dict]:

"""Busca erros recentes nos logs"""
 filter_str = f"""
severity >= ERROR
timestamp >= "{([Link]() - timedelta(hours=hours)).isoformat()}Z"
"""

errors = []
for entry in self.logging_client.list_entries(filter_=filter_str):
[Link]({
'timestamp': [Link](),
'severity': [Link],
'message': [Link]('message', str([Link])),
'resource': [Link]
})

return sorted(errors, key=lambda x: x['timestamp'], reverse=True)

# Implementação de CI/CD para dados

class DataPipelineCI:
"""CI/CD para pipelines de dados"""

def __init__(self, project_id: str):

self.project_id = project_id
self.bq_client = [Link]()

def validate_schema_changes(self,
dataset_id: str,
table_id: str,
new_schema: List[[Link]]) -> Tuple[bool, List[str]]:
"""Valida mudanças de schema"""

table_ref = f"{self.project_id}.{dataset_id}.{table_id}"

try:
table = self.bq_client.get_table(table_ref)
current_schema = [Link]

issues = []

# Verificar remoção de campos

current_fields = {[Link] for field in current_schema}
new_fields = {[Link] for field in new_schema}

removed_fields = current_fields - new_fields

if removed_fields:
[Link](f"Campos removidos: {removed_fields}")

# Verificar mudanças de tipo incompatíveis

 field_map = {[Link]: field for field in current_schema}

for new_field in new_schema:

if new_field.name in field_map:
current_field = field_map[new_field.name]

# Verificar mudança de tipo

if current_field.field_type != new_field.field_type:
if not self._is_compatible_type_change(
current_field.field_type,
new_field.field_type
):
[Link](
f"Mudança de tipo incompatível: "
f"{new_field.name} de {current_field.field_type} "
f"para {new_field.field_type}"
)

# Verificar mudança de modo

if current_field.mode == 'REQUIRED' and new_field.mode != 'REQUIRED':
[Link](
f"Campo obrigatório tornando-se opcional: {new_field.name}"
)

return len(issues) == 0, issues

except Exception as e:
# Tabela não existe ainda
return True, []

def _is_compatible_type_change(self, old_type: str, new_type: str) -> bool:

"""Verifica se mudança de tipo é compatível"""

compatible_changes = {
'INTEGER': ['NUMERIC', 'FLOAT', 'STRING'],
'NUMERIC': ['STRING'],
'FLOAT': ['STRING'],
'STRING': [], # String não pode mudar para outros tipos
'TIMESTAMP': ['STRING'],
'DATE': ['STRING', 'TIMESTAMP'],
'TIME': ['STRING'],
'DATETIME': ['STRING', 'TIMESTAMP']
}

return new_type in compatible_changes.get(old_type, [])

Projeto Capstone: Sistema Completo de Data Platform

Requisitos
Ingestão em tempo real de múltiplas fontes
Processamento com baixa latência

Analytics em tempo real

Machine Learning integrado
Dashboards interativos

Monitoramento completo

Implementação
[O código do projeto capstone seria extenso demais para incluir aqui, mas incluiria:]

1. Arquitetura de Microserviços com Cloud Run

2. Event Sourcing com Pub/Sub e Firestore
3. CQRS Pattern com BigQuery para leitura

4. Feature Store com Vertex AI

5. A/B Testing Framework

6. Data Lineage tracking

7. Automated Data Quality checks
8. Cost Optimization automation

Recursos Adicionais

Certificações Recomendadas
1. Google Cloud Professional Data Engineer
2. Google Cloud Professional Machine Learning Engineer

3. MongoDB Certified Developer

Próximos Passos
1. Especializações: Streaming (Apache Beam), ML Ops, Data Mesh
2. Linguagens: Go para performance, Rust para sistemas

3. Frameworks: dbt para transformações, Airflow para orquestração

4. Avançado: Kubernetes operators, Service mesh, Multi-cloud

Comunidades
Google Cloud Community
Python Brasil

DataOps Brasil
MLOps Community

Conclusão
Este curso fornece uma base sólida e prática para engenharia de dados moderna. A progressão foi desenhada
para construir competências incrementalmente, sempre com foco em projetos reais e aplicáveis.

Lembre-se: a excelência vem da prática constante e da curiosidade em explorar novas soluções. Continue
construindo, quebrando e reconstruindo.

"O código é poesia, os dados são a tinta, e a nuvem é nossa tela infinita."