from delta.tables import DeltaTable
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# SCD Type 2 -merge incremental com historico
w = Window.partitionBy("cliente_id")
    .orderBy(F.col("_ingested_at").desc())

df_dedup = (
  spark.readStream.format("delta")
    .option("ignoreChanges", "true")
    .load("bronze.vendas_raw")
    .withColumn("_rn", F.row_number().over(w))
    .filter("_rn = 1").drop("_rn")
)

target = DeltaTable.forName(spark, "silver.vendas")
target.alias("t").merge(
  df_dedup.alias("s"),
  "t.venda_id = s.venda_id"
).whenMatchedUpdate(
  condition="s._ingested_at > t._ingested_at",
  set={"*"}
).whenNotMatchedInsertAll().execute()

WITH base AS (
  SELECT v.regiao, v.canal,
    DATE_TRUNC('month', v.data_venda) AS mes,
    SUM(v.valor_liquido) AS receita,
    COUNT(DISTINCT v.cliente_id) AS clientes
  FROM silver.vendas v
  INNER JOIN silver.clientes c ON v.cliente_id = c.id
  WHERE v.status = 'concluida'
    AND c.churn_flag = 0
  GROUP BY 1, 2, 3
)
SELECT *,
  LAG(receita, 1) OVER(
    PARTITION BY regiao, canal ORDER BY mes
  ) AS receita_anterior,
  ROUND((receita - LAG(receita,1) OVER(
    PARTITION BY regiao, canal ORDER BY mes
  )) / NULLIF(LAG(receita,1) OVER(
    PARTITION BY regiao, canal ORDER BY mes
  ), 0), 4) AS crescimento_pct
FROM base

bundle:
  name: hmdata-pipeline-vendas

resources:
  pipelines:
    vendas_dlt:
      target: gold
      catalog: prod
      libraries:
        - notebook:
          path: ./src/silver_vendas_scd.py
        - notebook:
          path: ./src/gold_receita_regional.sql

{
  "source": "crm_salesforce",
  "destination": "bronze.leads",
  "mode": "incremental",
  "watermark": "updated_at",
  "quality_checks": [
    "not_null(email)",
    "unique(lead_id)"
  ]
}

$ databricks bundle deploy
# validating pipeline...
✓ vendas_dlt deployed
✓ daily_refresh scheduled
$ databricks bundle run

def validate_schema(df, rules):
  for rule in rules:
    result = rule.evaluate(df)
    if result.failed:
      alert("quality_breach",
        table=df.name,
        rows=result.count)
  return df.filter(valid)

SELECT cliente_id,
  DATEDIFF(now(), last_order)
    AS dias_inativo,
  predict_churn(features)
    AS churn_prob
FROM gold.clientes

services:
  api:
    build: ./api
    ports: ["8000:8000"]
    depends_on: [db, redis]
  worker:
    build: ./worker
    command: celery -A tasks