M08 — Pipeline fim a fim — case completo

                       ┌──────────────────────────────┐
                       │ Azure SQL  AdventureWorksLT  │
                       │   SalesLT.Customer            │
                       │   SalesLT.SalesOrderHeader    │
                       │   SalesLT.SalesOrderDetail    │
                       │   SalesLT.Product             │
                       └────────┬──────────────────────┘
                                │  (1) Copy diário (ADF)
                                ▼
                  ┌─────────────────────────────────────┐
                  │ ADLS Gen2 — BRONZE (parquet, daily) │
                  └────────┬────────────────────────────┘
                           │  (2) Notebook ingest
                           ▼
                  ┌──────────────────────────────────┐
                  │ silver_prd.adventureworks.*       │
                  │  - customer, orders, products     │
                  │  (Delta, tipado, dedup, MERGE)    │
                  └────────┬─────────────────────────┘
                           │  (3) Notebook agregação
                           ▼
                  ┌──────────────────────────────────┐
                  │ gold_prd.adventureworks.*         │
                  │  - fact_sales                     │
                  │  - dim_customer, dim_product, dim_date │
                  └────────┬─────────────────────────┘
                           │  (4) SQL Warehouse expõe
                           ▼
                  ┌─────────────────────┐
                  │ Power BI Dashboard  │
                  └─────────────────────┘

-- silver_prd.adventureworks.customer
CREATE SCHEMA IF NOT EXISTS silver_prd.adventureworks;

CREATE TABLE IF NOT EXISTS silver_prd.adventureworks.customer (
  customer_id INT,
  first_name STRING, last_name STRING,
  company_name STRING,
  email STRING, phone STRING,
  modified_at TIMESTAMP,
  _silver_ingested_at TIMESTAMP
) USING DELTA;

-- silver_prd.adventureworks.sales_order_header
CREATE TABLE IF NOT EXISTS silver_prd.adventureworks.sales_order_header (
  sales_order_id INT,
  customer_id INT,
  order_date DATE,
  ship_date DATE,
  status INT,
  subtotal DOUBLE, tax_amt DOUBLE, freight DOUBLE, total_due DOUBLE,
  modified_at TIMESTAMP,
  _silver_ingested_at TIMESTAMP
) USING DELTA PARTITIONED BY (order_date);

-- silver_prd.adventureworks.sales_order_detail
CREATE TABLE IF NOT EXISTS silver_prd.adventureworks.sales_order_detail (
  sales_order_id INT,
  sales_order_detail_id INT,
  product_id INT,
  order_qty INT,
  unit_price DOUBLE, line_total DOUBLE,
  modified_at TIMESTAMP,
  _silver_ingested_at TIMESTAMP
) USING DELTA;

-- silver_prd.adventureworks.product
CREATE TABLE IF NOT EXISTS silver_prd.adventureworks.product (
  product_id INT,
  name STRING, color STRING, size STRING,
  standard_cost DOUBLE, list_price DOUBLE,
  product_category STRING,
  modified_at TIMESTAMP,
  _silver_ingested_at TIMESTAMP
) USING DELTA;

CREATE SCHEMA IF NOT EXISTS gold_prd.adventureworks;

-- DIM CUSTOMER
CREATE TABLE IF NOT EXISTS gold_prd.adventureworks.dim_customer (
  customer_sk BIGINT GENERATED ALWAYS AS IDENTITY,
  customer_id INT,
  full_name STRING,
  company_name STRING,
  email STRING,
  _scd_valid_from TIMESTAMP, _scd_valid_to TIMESTAMP, _scd_current BOOLEAN
) USING DELTA;

-- DIM PRODUCT
CREATE TABLE IF NOT EXISTS gold_prd.adventureworks.dim_product (
  product_sk BIGINT GENERATED ALWAYS AS IDENTITY,
  product_id INT,
  name STRING, color STRING, product_category STRING,
  list_price DOUBLE
) USING DELTA;

-- DIM DATE
CREATE TABLE IF NOT EXISTS gold_prd.adventureworks.dim_date (
  date_key DATE,
  year INT, quarter INT, month INT, month_name STRING,
  day INT, weekday INT, is_weekend BOOLEAN
) USING DELTA;

-- FACT SALES
CREATE TABLE IF NOT EXISTS gold_prd.adventureworks.fact_sales (
  sales_order_id INT,
  sales_order_detail_id INT,
  date_key DATE,
  customer_sk BIGINT, product_sk BIGINT,
  order_qty INT,
  unit_price DOUBLE, line_total DOUBLE
) USING DELTA PARTITIONED BY (date_key);

# Databricks notebook source
from pyspark.sql import functions as F
from delta.tables import DeltaTable

dbutils.widgets.text("storage_account", "")
dbutils.widgets.text("catalog", "")
dbutils.widgets.text("data_processamento", "")

storage = dbutils.widgets.get("storage_account")
catalog = dbutils.widgets.get("catalog")
data_proc = dbutils.widgets.get("data_processamento")  # YYYY-MM-DD

# COMMAND ----------

# Credenciais (também já config no init script do cluster em prd)
for k in ["sp-appid", "sp-secret", "sp-tenant"]:
    pass  # já carregado via cluster config; aqui é só ilustrativo

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

# COMMAND ----------

def merge_bronze_to_silver(table_name: str, key_cols: list[str], select_expr: dict[str, str]):
    """Lê parquet do bronze do dia, transforma com select_expr, upsert no silver."""
    path = f"{bronze_root}/source=adventureworks/schema=salesLT/table={table_name}/ingestion_date={data_proc}/"
    df = spark.read.parquet(path)

    df = df.selectExpr(*[f"{v} AS {k}" for k, v in select_expr.items()]) \
           .withColumn("_silver_ingested_at", F.current_timestamp())

    target_tbl = f"{catalog}.adventureworks.{table_name}"
    if not spark.catalog.tableExists(target_tbl):
        df.write.format("delta").saveAsTable(target_tbl)
        return df.count(), 0

    target = DeltaTable.forName(spark, target_tbl)
    cond = " AND ".join([f"t.{c} = s.{c}" for c in key_cols])
    res = (
        target.alias("t")
        .merge(df.alias("s"), cond)
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )
    return df.count(), res

# COMMAND ----------

cust_total, _ = merge_bronze_to_silver(
    "customer",
    key_cols=["customer_id"],
    select_expr={
        "customer_id": "CAST(CustomerID AS INT)",
        "first_name":  "FirstName",
        "last_name":   "LastName",
        "company_name":"CompanyName",
        "email":       "EmailAddress",
        "phone":       "Phone",
        "modified_at": "CAST(ModifiedDate AS TIMESTAMP)",
    }
)
print(f"Customer: {cust_total} linhas processadas.")

# COMMAND ----------

merge_bronze_to_silver(
    "sales_order_header",
    key_cols=["sales_order_id"],
    select_expr={
        "sales_order_id": "CAST(SalesOrderID AS INT)",
        "customer_id":    "CAST(CustomerID AS INT)",
        "order_date":     "CAST(OrderDate AS DATE)",
        "ship_date":      "CAST(ShipDate AS DATE)",
        "status":         "CAST(Status AS INT)",
        "subtotal":       "CAST(SubTotal AS DOUBLE)",
        "tax_amt":        "CAST(TaxAmt AS DOUBLE)",
        "freight":        "CAST(Freight AS DOUBLE)",
        "total_due":      "CAST(TotalDue AS DOUBLE)",
        "modified_at":    "CAST(ModifiedDate AS TIMESTAMP)",
    }
)

merge_bronze_to_silver(
    "sales_order_detail",
    key_cols=["sales_order_id","sales_order_detail_id"],
    select_expr={
        "sales_order_id":        "CAST(SalesOrderID AS INT)",
        "sales_order_detail_id": "CAST(SalesOrderDetailID AS INT)",
        "product_id":            "CAST(ProductID AS INT)",
        "order_qty":             "CAST(OrderQty AS INT)",
        "unit_price":            "CAST(UnitPrice AS DOUBLE)",
        "line_total":            "CAST(LineTotal AS DOUBLE)",
        "modified_at":           "CAST(ModifiedDate AS TIMESTAMP)",
    }
)

merge_bronze_to_silver(
    "product",
    key_cols=["product_id"],
    select_expr={
        "product_id":       "CAST(ProductID AS INT)",
        "name":             "Name",
        "color":            "Color",
        "size":             "Size",
        "standard_cost":    "CAST(StandardCost AS DOUBLE)",
        "list_price":       "CAST(ListPrice AS DOUBLE)",
        "product_category": "ProductCategoryID",  # simplificado; em prod faz lookup
        "modified_at":      "CAST(ModifiedDate AS TIMESTAMP)",
    }
)

# COMMAND ----------

# Validação básica
display(spark.sql(f"""
  SELECT
    (SELECT COUNT(*) FROM {catalog}.adventureworks.customer) customers,
    (SELECT COUNT(*) FROM {catalog}.adventureworks.sales_order_header) orders,
    (SELECT COUNT(*) FROM {catalog}.adventureworks.sales_order_detail) order_items,
    (SELECT COUNT(*) FROM {catalog}.adventureworks.product) products
"""))

# Databricks notebook source
from pyspark.sql import functions as F

dbutils.widgets.text("silver_catalog", "silver_prd")
dbutils.widgets.text("gold_catalog", "gold_prd")
S = dbutils.widgets.get("silver_catalog")
G = dbutils.widgets.get("gold_catalog")

# COMMAND ----------
# MAGIC %md ## dim_date — regenera completa

spark.sql(f"""
  CREATE OR REPLACE TABLE {G}.adventureworks.dim_date AS
  WITH calendar 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)  AS year,
    QUARTER(date_key) AS quarter,
    MONTH(date_key) AS month,
    DATE_FORMAT(date_key, 'MMMM') AS month_name,
    DAY(date_key) AS day,
    DAYOFWEEK(date_key) AS weekday,
    DAYOFWEEK(date_key) IN (1,7) AS is_weekend
  FROM calendar
""")

# COMMAND ----------
# MAGIC %md ## dim_customer (SCD1 simplificado)

spark.sql(f"""
  MERGE INTO {G}.adventureworks.dim_customer t
  USING (
    SELECT customer_id,
           CONCAT_WS(' ', first_name, last_name) AS full_name,
           company_name, email,
           CURRENT_TIMESTAMP() AS _scd_valid_from,
           CAST(NULL AS TIMESTAMP) AS _scd_valid_to,
           TRUE AS _scd_current
    FROM {S}.adventureworks.customer
  ) s
  ON t.customer_id = s.customer_id
  WHEN MATCHED THEN UPDATE SET
    full_name = s.full_name, company_name = s.company_name, email = s.email
  WHEN NOT MATCHED THEN INSERT
    (customer_id, full_name, company_name, email, _scd_valid_from, _scd_valid_to, _scd_current)
    VALUES (s.customer_id, s.full_name, s.company_name, s.email, s._scd_valid_from, s._scd_valid_to, s._scd_current)
""")

# COMMAND ----------
# MAGIC %md ## dim_product

spark.sql(f"""
  MERGE INTO {G}.adventureworks.dim_product t
  USING (
    SELECT product_id, name, color, product_category, list_price
    FROM {S}.adventureworks.product
  ) s
  ON t.product_id = s.product_id
  WHEN MATCHED THEN UPDATE SET name=s.name, color=s.color, product_category=s.product_category, list_price=s.list_price
  WHEN NOT MATCHED THEN INSERT (product_id, name, color, product_category, list_price)
    VALUES (s.product_id, s.name, s.color, s.product_category, s.list_price)
""")

# COMMAND ----------
# MAGIC %md ## fact_sales (overwrite estratégico por partição "order_date")

spark.sql(f"""
  CREATE OR REPLACE TABLE {G}.adventureworks.fact_sales
  PARTITIONED BY (date_key)
  AS
  SELECT
    h.sales_order_id,
    d.sales_order_detail_id,
    h.order_date AS date_key,
    dc.customer_sk,
    dp.product_sk,
    d.order_qty, d.unit_price, d.line_total
  FROM {S}.adventureworks.sales_order_header h
  JOIN {S}.adventureworks.sales_order_detail d ON h.sales_order_id = d.sales_order_id
  JOIN {G}.adventureworks.dim_customer dc ON dc.customer_id = h.customer_id AND dc._scd_current
  JOIN {G}.adventureworks.dim_product dp ON dp.product_id = d.product_id
""")

# COMMAND ----------
# MAGIC %md ## OPTIMIZE + ZORDER (performance pro Power BI)

spark.sql(f"OPTIMIZE {G}.adventureworks.fact_sales ZORDER BY (customer_sk, product_sk)")

# COMMAND ----------
# MAGIC %md ## Validação: contagens + KPIs

display(spark.sql(f"""
  SELECT
    d.year, d.month,
    COUNT(*) trips,
    ROUND(SUM(f.line_total), 2) revenue
  FROM {G}.adventureworks.fact_sales f
  JOIN {G}.adventureworks.dim_date d ON f.date_key = d.date_key
  GROUP BY d.year, d.month
  ORDER BY 1,2
"""))

# Databricks notebook source
dbutils.widgets.text("catalog", "silver_prd")
C = dbutils.widgets.get("catalog")

checks = [
    ("customer sem email", f"SELECT COUNT(*) FROM {C}.adventureworks.customer WHERE email IS NULL", 0),
    ("orders com total negativo", f"SELECT COUNT(*) FROM {C}.adventureworks.sales_order_header WHERE total_due < 0", 0),
    ("detail sem header (orfão)", f"""
        SELECT COUNT(*) FROM {C}.adventureworks.sales_order_detail d
        LEFT JOIN {C}.adventureworks.sales_order_header h ON h.sales_order_id = d.sales_order_id
        WHERE h.sales_order_id IS NULL
    """, 0),
]

falhas = []
for name, sql, expected in checks:
    actual = spark.sql(sql).collect()[0][0]
    print(f"{name}: esperado <= {expected}, real = {actual}")
    if actual > expected:
        falhas.append((name, actual, expected))

if falhas:
    msg = "Data quality fail:\n" + "\n".join([f"- {n}: {a} > {e}" for n,a,e in falhas])
    dbutils.notebook.exit(msg)

print("✅ DQ OK")

resources:
  jobs:
    aw_diario_e2e:
      name: aw_diario_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 ]
        no_alert_for_skipped_runs: true

      max_concurrent_runs: 1
      timeout_seconds: 1800     # 30 min SLA

      tasks:
        - task_key: ingest_bronze_to_silver
          notebook_task:
            notebook_path: ../notebooks/10_ingest_bronze_to_silver.py
            base_parameters:
              storage_account: ${var.storage_account}
              catalog: ${var.silver_catalog}
              data_processamento: "{{job.start_time.iso_date}}"
          job_cluster_key: small

        - task_key: data_quality
          depends_on: [{ task_key: ingest_bronze_to_silver }]
          notebook_task:
            notebook_path: ../notebooks/30_data_quality.py
            base_parameters:
              catalog: ${var.silver_catalog}
          job_cluster_key: small

        - task_key: silver_to_gold
          depends_on: [{ task_key: data_quality }]
          notebook_task:
            notebook_path: ../notebooks/20_silver_to_gold.py
            base_parameters:
              silver_catalog: ${var.silver_catalog}
              gold_catalog: ${var.gold_catalog}
          job_cluster_key: small

      job_clusters:
        - job_cluster_key: small
          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
              ambiente: ${bundle.target}
              case: aw_e2e

Revenue = SUMX(fact_sales, fact_sales[order_qty] * fact_sales[unit_price])
Avg Ticket = DIVIDE(SUM(fact_sales[line_total]), DISTINCTCOUNT(fact_sales[sales_order_id]))
YoY Revenue % =
    VAR cur = [Revenue]
    VAR prev = CALCULATE([Revenue], SAMEPERIODLASTYEAR(dim_date[date_key]))
    RETURN DIVIDE(cur - prev, prev)

CREATE MONITORING FOR TABLE gold_prd.adventureworks.fact_sales
  PROFILE TYPE TIME_SERIES (
    TIMESTAMP_COL_EXPR = date_key,
    GRANULARITIES = (MONTH)
  )
  SCHEDULE EVERY 1 DAY AT 07:00 TIMEZONE 'America/Sao_Paulo';