Databricks Certified Machine Learning Professional

Spark ML ou nó único?

Dados que não cabem num nó (centenas de milhões de linhas) → Spark ML. Dados que cabem → scikit-learn/XGBoost (mais algoritmos, mais rápido de iterar). Inferência em tempo real costuma usar modelo leve de nó único servido em endpoint.

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator

idx = StringIndexer(inputCols=["emprego", "educacao"], outputCols=["emp_i", "edu_i"])
ohe = OneHotEncoder(inputCols=["emp_i", "edu_i"], outputCols=["emp_v", "edu_v"])
vec = VectorAssembler(inputCols=["renda", "valor", "emp_v", "edu_v"], outputCol="features")
gbt = GBTClassifier(labelCol="aprovado")
grid = ParamGridBuilder().addGrid(gbt.maxDepth, [4, 6]).build()
cv = CrossValidator(estimator=Pipeline(stages=[idx, ohe, vec, gbt]), estimatorParamMaps=grid,
                    evaluator=BinaryClassificationEvaluator(labelCol="aprovado"), numFolds=3)
modelo = cv.fit(treino)

Um modelo por grupo: pandas function API

def treina_loja(pdf: pd.DataFrame) -> pd.DataFrame:
    m = LinearRegression().fit(pdf[["promo", "dia"]], pdf["vendas"])
    return pd.DataFrame({"loja": [pdf.loja.iloc[0]], "coef": [m.coef_[0]]})

df.groupBy("loja").applyInPandas(treina_loja, schema="loja string, coef double")

Cada grupo vai para um executor → milhares de modelos pequenos em paralelo.

Tuning distribuído

• Optuna + MLflow: MlflowSparkStudy com MlflowStorage (experimento do MLflow) distribui os trials pelo cluster e registra cada um.
• Ray on Spark: setup_ray_cluster(...) + Ray Tune; bom para deep learning, workloads Python não-Spark e paralelismo fino.
• Ray x Spark: Spark brilha em processamento de dados tabulares; Ray em tarefas Python heterogêneas, RL, tuning e DL.

Escala

• Vertical (máquina maior/GPU): simples, sem overhead de comunicação, mas tem teto.
• Horizontal (mais nós): escala para dados enormes, com custo de rede.
• Paralelismo de dados: cada worker treina com uma fatia dos dados e sincroniza gradientes (modelo cabe na GPU).
• Paralelismo de modelo: o modelo é dividido entre dispositivos (modelo não cabe numa GPU).

MLflow avançado

with mlflow.start_run(run_name="tuning") as pai:
    for params in grade:
        with mlflow.start_run(nested=True):
            mlflow.log_params(params); mlflow.log_metric("cv_auc", avalia(params))

Modelo customizado com pré-processamento em tempo real: classe mlflow.pyfunc.PythonModel com load_context (carrega artefatos) e predict (calcula features a partir da requisição e chama o modelo).

Feature Store avançada

• Point-in-time: feature table de séries temporais (timeseries_columns) + FeatureLookup(..., timestamp_lookup_key="ts_evento") → cada exemplo de treino recebe o valor da feature como era naquele momento, sem vazar o futuro.
• Online: publique a feature table num online store (Lakebase) para lookups em milissegundos no serving.
• On-demand features: funções Python do UC calculadas na hora da requisição (ex.: distância entre o usuário e a loja), usadas igual no treino e no serving.
• Streaming: pipelines de streaming atualizam a feature table continuamente.

Deploy code (recomendado)

dev  ──PR──▶  staging (testes de integração com dados de staging)  ──merge──▶  prod
                                          │
            o MESMO código de treino roda em prod com dados de prod e registra o modelo

Componentes: Git + CI, bundles por ambiente, jobs de feature engineering/treino/validação/deploy, registro no UC com aliases, serving e monitoramento.

Mapa do ciclo de vida

Atividade	Recurso
Features	Feature Engineering no UC
Experimentos	MLflow tracking
Versionar/promover	Modelos no UC + aliases
Orquestrar	Lakeflow Jobs
Servir	Model Serving
Monitorar	Lakehouse Monitoring + inference tables
Infra como código	Declarative Automation Bundles

Para comparar com o modelo de produção sem que ele mude de baixo de você, referencie-o por alias (@champion), não pela "última versão".

Testes

• Unitário (dev/CI): funções puras de features e pré-processamento, com pytest.
• Integração (staging): o pipeline todo de ponta a ponta com dados reais de staging.
• Funções em módulos .py (importáveis) testam melhor que funções dentro de notebooks.
• Mudou só hiperparâmetros? Reexecute testes de treino, avaliação e deploy — feature engineering não mudou.

Bundles para ML

resources:
  experiments:
    churn_exp: { name: /Shared/churn-${bundle.target} }
  registered_models:
    churn: { catalog_name: ${var.catalogo}, schema_name: ml, name: churn }
  model_serving_endpoints:
    churn_ep:
      name: churn-${bundle.target}
      config:
        served_entities:
          - entity_name: ${var.catalogo}.ml.churn
            entity_version: "3"
            workload_size: Small
            scale_to_zero_enabled: true

Retreino automático

Monitor detecta drift/queda de métrica → alert → job de retreino → avaliação contra o champion no mesmo conjunto de validação com a métrica de negócio certa → se melhor, alias champion move para a nova versão. Escolha da métrica: se a decisão usa limiares de probabilidade (ex.: bloquear acima de 95%), a qualidade da probabilidade importa → log loss.

Lakehouse Monitoring

Perfil	Tabela	Uso
Snapshot	Estado completo a cada refresh	Tabelas de dimensão, cadastro
Time series	Coluna de tempo	Dados que chegam ao longo do tempo
Inference	Requests + predições (+ rótulos)	Qualidade de modelo e drift

Gera duas tabelas: profile metrics (estatísticas por janela/fatia) e drift metrics (testes: KS para numéricas, qui-quadrado para categóricas, distância de Wasserstein, PSI, JS). Baseline = tabela de referência ou janela anterior. Métricas customizadas (agregadas, derivadas, de drift) e slicing por colunas.

Saúde do endpoint

Latência (p50/p99), requisições/s, taxa de erro 4xx/5xx, CPU, memória e concorrência provisionada — na aba de métricas do endpoint e exportáveis.

Estratégias

Estratégia	Como	Risco
Blue-green	Novo ambiente completo; troca 100% do tráfego de uma vez (com rollback rápido)	Médio: todos os usuários migram juntos
Canário	Uma fração pequena do tráfego vai ao novo modelo e aumenta aos poucos	Baixo: problema afeta poucos
Shadow	Novo modelo recebe cópia do tráfego, resposta não é usada	Nenhum para o usuário

No Model Serving: dois served entities no mesmo endpoint com traffic split (ex.: 95/5 → 80/20 → 0/100), escala horizontal (scale-out) e route optimization para alta vazão.

PyFunc customizado

class Modelo(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        self.vocab = json.load(open(context.artifacts["vocab"]))
        self.m = joblib.load(context.artifacts["modelo"])
    def predict(self, context, model_input):
        return self.m.predict(prepara(model_input, self.vocab))

mlflow.pyfunc.log_model("model", python_model=Modelo(),
    artifacts={"vocab": "vocab.json", "modelo": "modelo.pkl"},
    registered_model_name="main.ml.classificador", input_example=exemplo)

Consultar

from mlflow.deployments import get_deploy_client
client = get_deploy_client("databricks")
client.predict(endpoint="churn", inputs={"dataframe_records": registros})

O SDK cuida da autenticação e do formato. Também dá para criar/atualizar endpoints (client.create_endpoint), pela REST API ou pela UI.