M09 — Case real: Uber Eats end-to-end em Azure

┌─────────────────────────────────────────────────────────────┐
│                   FONTES OPERACIONAIS                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  📱 App B2C    ──►  MongoDB Atlas      (mongodb_users)      │
│  🏢 CRM B2B    ──►  Azure SQL          (mssql_users)        │
│  🍔 Catálogo   ──►  MySQL              (mysql_restaurants,  │
│                                         mysql_ratings)      │
│  🛵 Motoristas ──►  Postgres           (postgres_drivers)   │
│  📦 Pedidos    ──►  Kafka/Event Hubs   (kafka_orders,       │
│                                         kafka_status)       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

                  ┌────────────────────────────────────────────┐
                  │              AZURE DATA FACTORY             │
                  │  trigger diário 06h BRT + trigger streaming │
                  └─────────┬──────────────────────────────┬────┘
                            │                              │
       ┌────────────────────┼──────────────────────────────┘
       ▼                    ▼
┌──────────────┐    ┌──────────────┐
│ Azure SQL    │    │ JSON uploads │ (mongo/mysql/postgres → ADLS direto)
│ (mssql_users)│    │              │
└──────┬───────┘    └──────┬───────┘
       │                    │
       └─────────┬──────────┘
                 │ Copy Activity (batch diário)
                 ▼
       ┌────────────────────┐         ┌─────────────────────┐
       │ ADLS Gen2 — Bronze │ ◄────── │ Event Hubs ──┐      │
       │  /source=<src>/    │         │  (streaming) │      │
       │  /table=<t>/       │         │  kafka_orders│      │
       │  /ing_date=<d>/    │         │  kafka_status│      │
       └─────────┬──────────┘         └──────────────┘      │
                 │                                          │
                 │ Notebook batch + Auto Loader streaming   │
                 ▼                                          │
       ┌────────────────────────────────────────────────────┘
       │  AZURE DATABRICKS — silver_dev.ubereats.*
       │   ┌─────────────────────────────────────┐
       │   │ users_unified (MSSQL ∪ MongoDB)      │
       │   │ restaurants                          │
       │   │ ratings                              │
       │   │ drivers                              │
       │   │ orders (MERGE incremental)           │
       │   │ order_status (streaming, append)     │
       │   └─────────────────────────────────────┘
       │
       ▼ Notebook agregação
       ┌────────────────────────────────────────────────────┐
       │  gold_dev.ubereats.*
       │   ┌──────────────────────────────────────────────┐
       │   │ dim_user (SCD2)                              │
       │   │ dim_restaurant (SCD1)                        │
       │   │ dim_driver                                   │
       │   │ dim_date                                     │
       │   │ dim_status                                   │
       │   │ fact_orders   (granularidade: pedido)        │
       │   │ fact_order_status (granularidade: evento)    │
       │   └──────────────────────────────────────────────┘
       │
       ▼ SQL Warehouse Pro Serverless
       ┌────────────────┐
       │   Power BI     │
       │   - SLA entrega│
       │   - GMV por    │
       │     restaurante│
       │   - Top motorista│
       └────────────────┘

"""
Lê kafka_orders.json e kafka_status.json da pasta entities/
e publica os eventos no Event Hubs simulando produção real.
"""
import json, os, time, random
from azure.eventhub import EventHubProducerClient, EventData

CONN_STR = os.environ["EH_CONN_STR"]  # cola do KV antes de rodar
MEC_PATH = r"C:\Users\willi\OneDrive\Documents\Projetos\Data Engineer\frm-databricks\frm-spark-databricks-mec\entities"

def send_to_hub(hub_name: str, jsonl_path: str, throttle_ms: int = 50):
    producer = EventHubProducerClient.from_connection_string(CONN_STR, eventhub_name=hub_name)
    batch = producer.create_batch()
    sent = 0
    with open(jsonl_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line: continue
            ev = EventData(line)
            try:
                batch.add(ev)
            except ValueError:
                producer.send_batch(batch)
                batch = producer.create_batch()
                batch.add(ev)
            sent += 1
            if throttle_ms: time.sleep(throttle_ms / 1000)
    if len(batch) > 0:
        producer.send_batch(batch)
    producer.close()
    print(f"{hub_name}: {sent} eventos enviados")

if __name__ == "__main__":
    send_to_hub("orders",       os.path.join(MEC_PATH, "kafka_orders.json"))
    send_to_hub("order-status", os.path.join(MEC_PATH, "kafka_status.json"))

CREATE SCHEMA IF NOT EXISTS silver_dev.ubereats
  COMMENT 'Camada Silver - case Uber Eats (M09)';

CREATE SCHEMA IF NOT EXISTS gold_dev.ubereats
  COMMENT 'Camada Gold - case Uber Eats (M09)';

# Databricks notebook source
# MAGIC %md # M09 — Bronze → Silver (Uber Eats batch)

# COMMAND ----------
from pyspark.sql import functions as F
from pyspark.sql.types import *
from delta.tables import DeltaTable

dbutils.widgets.text("storage_account", "")
dbutils.widgets.text("ingestion_date", "")
storage = dbutils.widgets.get("storage_account")
ing_date = dbutils.widgets.get("ingestion_date")

bronze_root = f"abfss://bronze@{storage}.dfs.core.windows.net"

# Helper genérico
def merge_jsonl_to_silver(source: str, table: str, key_cols: list[str], select_expr: dict[str, str]):
    path = f"{bronze_root}/source={source}/table={table}/ingestion_date={ing_date}/"
    df = spark.read.json(path)
    df = df.selectExpr(*[f"{v} AS {k}" for k, v in select_expr.items()]) \
           .withColumn("_silver_ingested_at", F.current_timestamp()) \
           .withColumn("_silver_source", F.lit(source))

    target = f"silver_dev.ubereats.{table}"
    if not spark.catalog.tableExists(target):
        df.write.format("delta").saveAsTable(target)
        return df.count(), 0
    t = DeltaTable.forName(spark, target)
    cond = " AND ".join([f"t.{c} = s.{c}" for c in key_cols])
    t.alias("t").merge(df.alias("s"), cond) \
        .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
    return df.count(), -1

# COMMAND ----------
# MAGIC %md ## 3.1 — restaurants (MySQL → silver)

merge_jsonl_to_silver(
    source="mysql", table="restaurants",
    key_cols=["restaurant_id"],
    select_expr={
        "restaurant_id":  "CAST(restaurant_id AS INT)",
        "cnpj":           "TRIM(cnpj)",
        "name":           "name",
        "cuisine_type":   "cuisine_type",
        "city":           "city",
        "country":        "country",
        "phone_number":   "phone_number",
        "address":        "address",
        "opening_time":   "opening_time",
        "closing_time":   "closing_time",
        "average_rating": "CAST(average_rating AS DOUBLE)",
        "num_reviews":    "CAST(num_reviews AS INT)",
        "uuid":           "uuid",
        "_src_ts":        "CAST(dt_current_timestamp AS TIMESTAMP)",
    }
)

# COMMAND ----------
# MAGIC %md ## 3.2 — ratings (MySQL → silver)

merge_jsonl_to_silver(
    source="mysql", table="ratings",
    key_cols=["rating_id"],
    select_expr={
        "rating_id":             "CAST(rating_id AS INT)",
        "restaurant_cnpj":       "TRIM(restaurant_identifier)",
        "rating":                "CAST(rating AS INT)",
        "rated_at":              "CAST(timestamp AS TIMESTAMP)",
        "uuid":                  "uuid",
        "_src_ts":               "CAST(dt_current_timestamp AS TIMESTAMP)",
    }
)

# COMMAND ----------
# MAGIC %md ## 3.3 — drivers (Postgres → silver)

merge_jsonl_to_silver(
    source="postgres", table="drivers",
    key_cols=["driver_id"],
    select_expr={
        "driver_id":             "CAST(driver_id AS INT)",
        "license_number":        "TRIM(license_number)",
        "first_name":            "first_name",
        "last_name":             "last_name",
        "phone_number":          "phone_number",
        "city":                  "city",
        "country":               "country",
        "date_birth":            "CAST(date_birth AS DATE)",
        "vehicle_type":          "vehicle_type",
        "vehicle_make":          "vehicle_make",
        "vehicle_model":         "vehicle_model",
        "vehicle_year":          "CAST(vehicle_year AS INT)",
        "vehicle_license_plate": "vehicle_license_plate",
        "uuid":                  "uuid",
        "_src_ts":               "CAST(dt_current_timestamp AS TIMESTAMP)",
    }
)

# COMMAND ----------
# MAGIC %md ## 3.4 — users B2C (MongoDB) e B2B (MSSQL) → silver unificado
# MAGIC
# MAGIC Desafio: o mesmo CPF pode existir em ambos. Estratégia:
# MAGIC 1. Gravar em duas tabelas separadas (`users_b2c`, `users_b2b`) primeiro.
# MAGIC 2. Construir uma view unificada `users_unified` que prioriza MSSQL (mais completo: birthday, job, company).
# MAGIC 3. Marca origem em `_source_priority`.

merge_jsonl_to_silver(
    source="mongodb", table="users_b2c",
    key_cols=["user_id"],
    select_expr={
        "user_id":          "CAST(user_id AS INT)",
        "cpf":              "TRIM(user_identifier)",
        "email":            "LOWER(TRIM(email))",
        "phone_number":     "phone_number",
        "city":             "city",
        "country":          "country",
        "delivery_address": "delivery_address",
        "uuid":             "uuid",
        "_src_ts":          "CAST(dt_current_timestamp AS TIMESTAMP)",
    }
)

merge_jsonl_to_silver(
    source="mssql", table="users_b2b",
    key_cols=["user_id"],
    select_expr={
        "user_id":      "CAST(user_id AS INT)",
        "cpf":          "TRIM(cpf)",
        "first_name":   "first_name",
        "last_name":    "last_name",
        "company_name": "company_name",
        "job":          "job",
        "birthday":     "CAST(birthday AS DATE)",
        "phone_number": "phone_number",
        "country":      "country",
        "uuid":         "uuid",
        "_src_ts":      "CAST(dt_current_timestamp AS TIMESTAMP)",
    }
)

# COMMAND ----------
# MAGIC %md ### users_unified
# MAGIC Tabela canônica resolvendo o mesmo CPF em ambas as fontes. Chave: cpf.

spark.sql("""
CREATE OR REPLACE TABLE silver_dev.ubereats.users_unified
USING DELTA AS
WITH b2b AS (
  SELECT cpf, first_name, last_name, company_name, job, birthday, phone_number, country,
         CAST(NULL AS STRING) AS email, CAST(NULL AS STRING) AS city, CAST(NULL AS STRING) AS delivery_address,
         'b2b' AS source_priority, _src_ts
  FROM silver_dev.ubereats.users_b2b
),
b2c AS (
  SELECT cpf,
         CAST(NULL AS STRING) AS first_name,
         CAST(NULL AS STRING) AS last_name,
         CAST(NULL AS STRING) AS company_name,
         CAST(NULL AS STRING) AS job,
         CAST(NULL AS DATE)   AS birthday,
         phone_number, country, email, city, delivery_address,
         'b2c' AS source_priority, _src_ts
  FROM silver_dev.ubereats.users_b2c
),
unioned AS (
  SELECT * FROM b2b
  UNION ALL
  SELECT * FROM b2c
),
ranked AS (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY cpf
      ORDER BY CASE source_priority WHEN 'b2b' THEN 1 WHEN 'b2c' THEN 2 END, _src_ts DESC
    ) AS _rn
  FROM unioned
)
SELECT cpf, first_name, last_name, company_name, job, birthday,
       phone_number, country, email, city, delivery_address,
       source_priority, _src_ts AS source_updated_at,
       CURRENT_TIMESTAMP() AS _silver_built_at
FROM ranked
WHERE _rn = 1
""")

display(spark.sql("""
  SELECT source_priority, COUNT(*) qtd
  FROM silver_dev.ubereats.users_unified
  GROUP BY source_priority
"""))

# COMMAND ----------
# MAGIC %md ## 3.5 — orders (Event Hubs batch read)

# Lê eventos acumulados do EH via Spark (não streaming aqui)
eh_conn = dbutils.secrets.get("kv", "eh-connstr")
eh_conf = {
    "eventhubs.connectionString": spark._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(eh_conn),
    "eventhubs.consumerGroup": "$Default",
    "eventhubs.startingPosition": '{"offset":"-1","seqNo":-1,"enqueuedTime":null,"isInclusive":true}'
}

df_raw_orders = (
    spark.read.format("eventhubs")
    .options(**{**eh_conf, "eventhubs.name": "orders"})
    .load()
)

# Payload do EH vem em "body" (binary). Decode + parse JSON.
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_key", StringType()),
    StructField("restaurant_key", StringType()),
    StructField("driver_key", StringType()),
    StructField("order_date", StringType()),
    StructField("total_amount", DoubleType()),
    StructField("payment_id", StringType()),
    StructField("dt_current_timestamp", StringType()),
])

orders = (
    df_raw_orders
    .select(F.from_json(F.col("body").cast("string"), order_schema).alias("o"),
            F.col("enqueuedTime").alias("_eh_ts"))
    .select("o.*", "_eh_ts")
    .withColumn("order_date",  F.to_timestamp("order_date"))
    .withColumn("user_cpf",    F.trim("user_key"))
    .withColumn("restaurant_cnpj", F.trim("restaurant_key"))
    .withColumn("driver_license",  F.trim("driver_key"))
    .withColumn("_silver_ingested_at", F.current_timestamp())
    .drop("user_key", "restaurant_key", "driver_key")
)

# MERGE
target = "silver_dev.ubereats.orders"
if not spark.catalog.tableExists(target):
    orders.write.format("delta").partitionBy(F.to_date("order_date")).saveAsTable(target)
else:
    t = DeltaTable.forName(spark, target)
    t.alias("t").merge(orders.alias("s"), "t.order_id = s.order_id") \
        .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

print(f"orders: {orders.count()} eventos processados")

# COMMAND ----------
display(spark.table("silver_dev.ubereats.orders").orderBy(F.desc("order_date")).limit(5))

# Databricks notebook source
# MAGIC %md # M09 — Streaming order_status (Event Hubs → Silver Delta)
# MAGIC
# MAGIC Diferente do batch: rodamos em modo streaming porque order_status muda em tempo real.
# MAGIC Schedule: `continuous` (trigger=AvailableNow para teste; Trigger.ProcessingTime("30 seconds") em prod).

# COMMAND ----------
from pyspark.sql import functions as F
from pyspark.sql.types import *

storage = dbutils.widgets.get("storage_account") if "storage_account" in [w.name for w in dbutils.widgets.getAll()] else None
checkpoint_root = f"abfss://silver@{storage}.dfs.core.windows.net/_checkpoints/order_status"
output_path     = f"abfss://silver@{storage}.dfs.core.windows.net/ubereats/order_status"

eh_conn = dbutils.secrets.get("kv", "eh-connstr")
eh_conf = {
    "eventhubs.connectionString": spark._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(eh_conn),
    "eventhubs.consumerGroup": "$Default",
    "eventhubs.startingPosition": '{"offset":"-1","seqNo":-1,"enqueuedTime":null,"isInclusive":true}',
    "eventhubs.name": "order-status",
}

status_schema = StructType([
    StructField("status_id", IntegerType()),
    StructField("order_identifier", StringType()),
    StructField("status", StructType([
        StructField("status_name", StringType()),
        StructField("timestamp", DoubleType()),  # epoch ms
    ])),
    StructField("dt_current_timestamp", StringType()),
])

stream = (
    spark.readStream.format("eventhubs").options(**eh_conf).load()
    .select(F.from_json(F.col("body").cast("string"), status_schema).alias("s"),
            F.col("enqueuedTime").alias("_eh_ts"))
    .select(
        F.col("s.status_id").alias("status_id"),
        F.col("s.order_identifier").alias("order_id"),
        F.col("s.status.status_name").alias("status_name"),
        F.from_unixtime(F.col("s.status.timestamp") / 1000).cast("timestamp").alias("status_ts"),
        F.col("s.dt_current_timestamp").cast("timestamp").alias("_src_ts"),
        F.col("_eh_ts"),
        F.current_timestamp().alias("_silver_ingested_at"),
    )
)

(
    stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", checkpoint_root)
    .option("path", output_path)
    .partitionBy(F.to_date("status_ts"))
    .trigger(availableNow=True)   # processa tudo que está acumulado e PARA. Em prod: .trigger(processingTime="30 seconds")
    .toTable("silver_dev.ubereats.order_status")
)

# COMMAND ----------
display(spark.table("silver_dev.ubereats.order_status").orderBy(F.desc("status_ts")).limit(10))

# Databricks notebook source
# MAGIC %md # M09 — Silver → Gold (star schema Uber Eats)

# COMMAND ----------
from pyspark.sql import functions as F

# COMMAND ----------
# MAGIC %md ## dim_date

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.dim_date AS
WITH cal AS (
  SELECT explode(sequence(to_date('2020-01-01'), to_date('2030-12-31'), interval 1 day)) AS date_key
)
SELECT date_key,
  YEAR(date_key) y, QUARTER(date_key) q, MONTH(date_key) m,
  DATE_FORMAT(date_key, 'MMMM') month_name,
  DAY(date_key) d, DAYOFWEEK(date_key) dow,
  DAYOFWEEK(date_key) IN (1,7) AS is_weekend
FROM cal
""")

# COMMAND ----------
# MAGIC %md ## dim_user (SCD2 minimal — capta mudança de email/endereço)

spark.sql("""
CREATE TABLE IF NOT EXISTS gold_dev.ubereats.dim_user (
  user_sk BIGINT GENERATED ALWAYS AS IDENTITY,
  cpf STRING, first_name STRING, last_name STRING, company_name STRING,
  email STRING, city STRING, country STRING, delivery_address STRING,
  source_priority STRING,
  _scd_valid_from TIMESTAMP, _scd_valid_to TIMESTAMP, _scd_current BOOLEAN
) USING DELTA
""")

# Carga inicial (idempotente: substitui dim atual pela versão mais recente de cada CPF)
spark.sql("""
MERGE INTO gold_dev.ubereats.dim_user AS t
USING (
  SELECT cpf, first_name, last_name, company_name, email, city, country, delivery_address, source_priority,
         CURRENT_TIMESTAMP() AS _scd_valid_from,
         CAST(NULL AS TIMESTAMP) AS _scd_valid_to,
         TRUE AS _scd_current
  FROM silver_dev.ubereats.users_unified
) s
ON t.cpf = s.cpf AND t._scd_current = TRUE
WHEN MATCHED AND (
     COALESCE(t.email,'')            <> COALESCE(s.email,'')
  OR COALESCE(t.delivery_address,'') <> COALESCE(s.delivery_address,'')
) THEN UPDATE SET _scd_valid_to = CURRENT_TIMESTAMP(), _scd_current = FALSE
WHEN NOT MATCHED THEN INSERT
  (cpf, first_name, last_name, company_name, email, city, country, delivery_address, source_priority,
   _scd_valid_from, _scd_valid_to, _scd_current)
  VALUES (s.cpf, s.first_name, s.last_name, s.company_name, s.email, s.city, s.country, s.delivery_address, s.source_priority,
          s._scd_valid_from, s._scd_valid_to, s._scd_current)
""")

# Pós-MERGE: insere a nova versão dos que tiveram mudança
spark.sql("""
INSERT INTO gold_dev.ubereats.dim_user
  (cpf, first_name, last_name, company_name, email, city, country, delivery_address, source_priority,
   _scd_valid_from, _scd_valid_to, _scd_current)
SELECT s.cpf, s.first_name, s.last_name, s.company_name, s.email, s.city, s.country, s.delivery_address, s.source_priority,
       CURRENT_TIMESTAMP(), NULL, TRUE
FROM silver_dev.ubereats.users_unified s
JOIN gold_dev.ubereats.dim_user t ON t.cpf = s.cpf AND t._scd_current = FALSE AND t._scd_valid_to >= CURRENT_TIMESTAMP() - INTERVAL 1 MINUTE
WHERE NOT EXISTS (
  SELECT 1 FROM gold_dev.ubereats.dim_user t2
  WHERE t2.cpf = s.cpf AND t2._scd_current = TRUE
)
""")

# COMMAND ----------
# MAGIC %md ## dim_restaurant (SCD1 — sobrescreve)

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.dim_restaurant AS
SELECT
  ROW_NUMBER() OVER (ORDER BY restaurant_id) AS restaurant_sk,
  restaurant_id, cnpj, name, cuisine_type, city, country,
  average_rating, num_reviews,
  opening_time, closing_time
FROM silver_dev.ubereats.restaurants
""")

# COMMAND ----------
# MAGIC %md ## dim_driver (SCD1)

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.dim_driver AS
SELECT
  ROW_NUMBER() OVER (ORDER BY driver_id) AS driver_sk,
  driver_id, license_number,
  CONCAT_WS(' ', first_name, last_name) AS full_name,
  city, country,
  vehicle_type, vehicle_make, vehicle_model, vehicle_year, vehicle_license_plate
FROM silver_dev.ubereats.drivers
""")

# COMMAND ----------
# MAGIC %md ## dim_status (catálogo enxuto, do enum)

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.dim_status AS
SELECT * FROM VALUES
  (1, 'Order Placed',     'placed',    1),
  (2, 'In Analysis',      'analysis',  2),
  (3, 'Accepted',          'accepted',  3),
  (4, 'Preparing',         'preparing', 4),
  (5, 'Ready for Pickup',  'ready',     5),
  (6, 'Picked Up',         'picked',    6),
  (7, 'Delivered',         'delivered', 7),
  (8, 'Cancelled',         'cancelled', 99)
AS status_dim(status_sk, status_name, status_code, status_order)
""")

# COMMAND ----------
# MAGIC %md ## fact_orders (granularidade: pedido)

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.fact_orders
PARTITIONED BY (order_date_key)
AS
SELECT
  o.order_id,
  CAST(o.order_date AS DATE)         AS order_date_key,
  CAST(o.order_date AS TIMESTAMP)    AS order_ts,
  du.user_sk,
  dr.restaurant_sk,
  dd.driver_sk,
  o.total_amount,
  o.payment_id
FROM silver_dev.ubereats.orders o
LEFT JOIN gold_dev.ubereats.dim_user       du ON du.cpf            = o.user_cpf AND du._scd_current
LEFT JOIN gold_dev.ubereats.dim_restaurant dr ON dr.cnpj           = o.restaurant_cnpj
LEFT JOIN gold_dev.ubereats.dim_driver     dd ON dd.license_number = o.driver_license
""")

# COMMAND ----------
# MAGIC %md ## fact_order_status (granularidade: evento de status)

spark.sql("""
CREATE OR REPLACE TABLE gold_dev.ubereats.fact_order_status
PARTITIONED BY (status_date_key)
AS
SELECT
  os.order_id,
  CAST(os.status_ts AS DATE) AS status_date_key,
  os.status_ts,
  ds.status_sk,
  os.status_name
FROM silver_dev.ubereats.order_status os
LEFT JOIN gold_dev.ubereats.dim_status ds ON ds.status_name = os.status_name
""")

# COMMAND ----------
# MAGIC %md ## OPTIMIZE + ZORDER

spark.sql("OPTIMIZE gold_dev.ubereats.fact_orders ZORDER BY (user_sk, restaurant_sk)")
spark.sql("OPTIMIZE gold_dev.ubereats.fact_order_status ZORDER BY (order_id)")

# COMMAND ----------
# MAGIC %md ## KPIs sanity

display(spark.sql("""
  SELECT d.y AS year, d.m AS month, COUNT(*) AS orders,
         ROUND(SUM(f.total_amount), 2) AS gmv,
         ROUND(AVG(f.total_amount), 2) AS avg_ticket
  FROM gold_dev.ubereats.fact_orders f
  JOIN gold_dev.ubereats.dim_date d ON d.date_key = f.order_date_key
  GROUP BY d.y, d.m
  ORDER BY 1, 2
"""))

display(spark.sql("""
  SELECT r.cuisine_type, COUNT(*) AS orders, ROUND(SUM(f.total_amount), 2) AS gmv
  FROM gold_dev.ubereats.fact_orders f
  JOIN gold_dev.ubereats.dim_restaurant r ON r.restaurant_sk = f.restaurant_sk
  GROUP BY r.cuisine_type
  ORDER BY gmv DESC
"""))

# Databricks notebook source
checks = [
    ("orders.user_cpf nulo (FK quebrada)",
     "SELECT COUNT(*) FROM silver_dev.ubereats.orders WHERE user_cpf IS NULL OR user_cpf = ''", 0),

    ("orders sem user correspondente em users_unified",
     """SELECT COUNT(*) FROM silver_dev.ubereats.orders o
        LEFT JOIN silver_dev.ubereats.users_unified u ON u.cpf = o.user_cpf
        WHERE u.cpf IS NULL""", 0),

    ("orders sem restaurante correspondente",
     """SELECT COUNT(*) FROM silver_dev.ubereats.orders o
        LEFT JOIN silver_dev.ubereats.restaurants r ON r.cnpj = o.restaurant_cnpj
        WHERE r.cnpj IS NULL""", 0),

    ("orders com total_amount negativo",
     "SELECT COUNT(*) FROM silver_dev.ubereats.orders WHERE total_amount < 0", 0),

    ("ratings fora do range 1-5",
     "SELECT COUNT(*) FROM silver_dev.ubereats.ratings WHERE rating < 1 OR rating > 5", 0),

    ("users_unified duplicados por CPF (deveria ser 0)",
     "SELECT COUNT(*) FROM (SELECT cpf FROM silver_dev.ubereats.users_unified GROUP BY cpf HAVING COUNT(*) > 1)", 0),
]

falhas = []
for name, sql, threshold in checks:
    actual = spark.sql(sql).collect()[0][0]
    status = "OK" if actual <= threshold else "FAIL"
    print(f"[{status}] {name}: real={actual} (limite {threshold})")
    if status == "FAIL":
        falhas.append((name, actual, threshold))

if falhas:
    msg = "DQ FAIL:\n" + "\n".join([f"- {n}: {a} > {t}" for n,a,t in falhas])
    dbutils.notebook.exit(msg)

print("✅ DQ OK")

resources:
  jobs:
    ubereats_e2e:
      name: ubereats_e2e
      schedule:
        quartz_cron_expression: "0 0 9 * * ?"   # 06h BRT
        timezone_id: UTC
        pause_status: ${var.bundle_target == "prd" ? "UNPAUSED" : "PAUSED"}

      email_notifications:
        on_failure: [ team.old.schoolll@gmail.com ]

      max_concurrent_runs: 1
      timeout_seconds: 2400   # 40 min SLA

      tasks:
        - task_key: bronze_to_silver_batch
          notebook_task:
            notebook_path: ../notebooks/M09/10_ubereats_bronze_to_silver.py
            base_parameters:
              storage_account: ${var.storage_account}
              ingestion_date: "{{job.start_time.iso_date}}"
          job_cluster_key: ue_cluster

        - task_key: status_streaming_micro_batch
          depends_on: [{ task_key: bronze_to_silver_batch }]
          notebook_task:
            notebook_path: ../notebooks/M09/20_ubereats_status_streaming.py
            base_parameters:
              storage_account: ${var.storage_account}
          job_cluster_key: ue_cluster

        - task_key: dq_silver
          depends_on: [{ task_key: status_streaming_micro_batch }]
          notebook_task:
            notebook_path: ../notebooks/M09/40_ubereats_dq.py
          job_cluster_key: ue_cluster

        - task_key: silver_to_gold
          depends_on: [{ task_key: dq_silver }]
          notebook_task:
            notebook_path: ../notebooks/M09/30_ubereats_silver_to_gold.py
          job_cluster_key: ue_cluster

      job_clusters:
        - job_cluster_key: ue_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: Standard_DS3_v2
            num_workers: 0
            data_security_mode: SINGLE_USER
            spark_conf:
              spark.databricks.cluster.profile: singleNode
              spark.master: local[*, 4]
            custom_tags:
              projeto: de
              case: ubereats
              ambiente: ${bundle.target}

GMV         = SUM(fact_orders[total_amount])
Orders      = DISTINCTCOUNT(fact_orders[order_id])
Avg Ticket  = DIVIDE([GMV], [Orders])

Time to Delivery (min) =
    VAR placed   = CALCULATE(MIN(fact_order_status[status_ts]), fact_order_status[status_name] = "Order Placed")
    VAR picked   = CALCULATE(MIN(fact_order_status[status_ts]), fact_order_status[status_name] = "Picked Up")
    RETURN DIVIDE(DATEDIFF(placed, picked, MINUTE), 1)

SLA Compliance % =
    DIVIDE(
      CALCULATE([Orders], [Time to Delivery (min)] <= 45),
      [Orders]
    )