%pip install -q "jupyter-notebook-slide @ git+https://github.com/dsevilla/jupyter-notebook-slide.git"
%load_ext notebook_slide

import notebook_slide as jnbs

# Colores de 26-27/teoria/tcdm.css, adaptados al tema de las sesiones.
jnbs.configure(
    background="#eaf2f8",
    foreground="#1f2933",
    border="#c9d6e1",
    heading="#0c304d",
    subheading="#0c304d",
    link="#174f7a",
    code_background="#eaf2f8",
    code_foreground="#0c304d",
    quote_background="#ffffff",
    font_family="Atkinson Hyperlegible, Inter, Aptos, Segoe UI, Helvetica, Arial, sans-serif",
    font_url="https://fonts.googleapis.com/css2?family=Atkinson+Hyperlegible:ital,wght@0,400;0,700;1,400;1,700&display=swap",
)
jnbs.register_slide_type("avance", "A continuación", "#174f7a")
jnbs.register_slide_type("resumen", "Recapitulación", "#a54467")
jnbs.register_slide_type("pregunta", "Pregunta guía", "#0c304d")
jnbs.register_slide_type("evaluacion", "Evaluación de la sesión", "#b3701a")
jnbs.register_slide_type(
    "titulo",
    "Sección",
    "#174f7a",
    layout="title",
    background="linear-gradient(135deg, #0c304d, #174f7a 62%, #a54467)",
    foreground="#ffffff",
    border="transparent",
    heading="#ffffff",
    subheading="#dceaf4",
)
jnbs.register_alias("diapositiva")

S2: Parquet como ficheros
  → S3: los mismos Parquet en un almacenamiento de objetos
  → S4: Spark procesa por lotes
  → S5: zonas raw/silver/gold y particionado físico
  → S6: catálogo Hive y Trino
  → S7: tabla Iceberg, evolución y optimización medida
  → S8: la misma tabla Iceberg, alimentada de forma continua

cd ~/tcdm-public
git pull
make -C entorno warehouse-up

!hdfs dfs -ls -d \
    /datalake/raw/tpcds/web_sales \
    /datalake/raw/tpcds/customer \
    /datalake/raw/tpcds/customer_address \
    /datalake/raw/tpcds/item \
    /datalake/raw/tpcds/date_dim \
    /warehouse

%%writefile requirements.txt
# Dependencias del kernel interactivo de la sesión 8 (Spark local con un
# catálogo Iceberg de tipo hive, el mismo Hive Metastore que usa Trino). La
# serie de PySpark aceptada por el curso es >4,<4.2, igual que en S4, S5, S6
# y S7.
pyspark>4,<4.2
# Cliente DB-API 2.0 de Trino: se usa en las celdas de sólo lectura como
# comprobación independiente de lo que Spark escribe directamente en
# iceberg.tcdm.* dentro de foreachBatch (ver "Cómo configuramos el catálogo
# Iceberg de Spark" en el notebook). Se fija la serie mayor para que el
# tipado de columnas y la representación de valores no cambien entre
# ejecuciones.
trino>=0.330,<1
pandas
# generate_web_sales_batch.py habla con WebHDFS por fsspec y lee/escribe
# Parquet con PyArrow, igual que los lectores de la sesión 2; requests es
# la dependencia HTTP que usa fsspec para WebHDFS.
fsspec>=2026.2.0
pyarrow
requests

%pip install -q -r requirements.txt

import sys

import pandas
import pyspark
import trino

print(f"PySpark {pyspark.__version__} con Python {sys.version.split()[0]}.")
print(f"Cliente trino: {trino.__version__}")
print(f"Pandas: {pandas.__version__}")

from pyspark import SparkContext
from pyspark.sql import SparkSession

ICEBERG_VERSION: str = "1.11.0"
ICEBERG_RUNTIME_PACKAGE: str = (
    f"org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:{ICEBERG_VERSION}"
)
HIVE_METASTORE_URIS: str = "thrift://hive-metastore:9083"
WAREHOUSE_DIR: str = "hdfs://namenode:9000/warehouse"

spark: SparkSession = (
    SparkSession.builder.appName("TCDM_S8_Streaming")
    .master("local[2]")
    .config("spark.driver.memory", "1g")
    .config("spark.sql.shuffle.partitions", "8")
    .config("spark.jars.packages", ICEBERG_RUNTIME_PACKAGE)
    .config(
        "spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
    )
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.iceberg.type", "hive")
    .config("spark.sql.catalog.iceberg.uri", HIVE_METASTORE_URIS)
    .config("spark.sql.catalog.iceberg.warehouse", WAREHOUSE_DIR)
    .getOrCreate()
)
sc: SparkContext = spark.sparkContext

print(f"Spark version: {spark.version} | Master: {sc.master}")
print(f"Paquete Iceberg: {ICEBERG_RUNTIME_PACKAGE}")
print(f"Catálogo Iceberg: iceberg (Hive Metastore {HIVE_METASTORE_URIS})")

import subprocess

import pandas as pd
import trino
from trino.dbapi import Connection, Cursor

trino_conn: Connection = trino.dbapi.connect(
    host="trino-hdfs",
    port=8080,
    user="luser",
    catalog="hive",
    schema="tcdm",
)


def run_trino(sql: str) -> pd.DataFrame:
    '''Ejecuta una sentencia en Trino y devuelve el resultado como DataFrame.'''
    cursor: Cursor = trino_conn.cursor()
    cursor.execute(sql)
    rows: list[tuple[object, ...]] = cursor.fetchall()
    columns: list[str] = [description[0] for description in cursor.description or []]
    return pd.DataFrame(rows, columns=columns)


print(f"Cliente trino: {trino.__version__}")
run_trino("SELECT 1 AS ping")

def described_columns(table_name: str) -> dict[str, str]:
    '''Convertir el esquema de Spark de una tabla en un diccionario columna -> tipo.'''
    assert spark.catalog.tableExists(
        table_name
    ), f"No existe {table_name}. ¿Se ha ejecutado S7 antes que esta sesión?"
    return {field.name: field.dataType.simpleString() for field in spark.table(table_name).schema}


web_sales_columns: dict[str, str] = described_columns("iceberg.tcdm.web_sales")
for name, data_type in web_sales_columns.items():
    print(f"{name}: {data_type}")

expected_web_sales_columns: set[str] = {
    "ws_order_number",
    "ws_item_sk",
    "ws_bill_customer_sk",
    "ws_bill_addr_sk",
    "ws_sold_date_sk",
    "ws_sold_date",
    "ws_quantity",
    "ws_list_price",
    "ws_ext_discount_amt",
    "ws_ext_sales_price",
    "ws_net_paid",
    "ws_net_profit",
    "sold_year",
    "sold_month",
}
missing_web_sales_columns: set[str] = expected_web_sales_columns - set(web_sales_columns)
assert not missing_web_sales_columns, (
    "Faltan columnas en iceberg.tcdm.web_sales: "
    f"{missing_web_sales_columns}. ¿Se ha ejecutado S7 antes que esta sesión?"
)
print("iceberg.tcdm.web_sales tiene todas las columnas del contrato de S8.")

web_sales_by_year_columns: dict[str, str] = described_columns("iceberg.tcdm.web_sales_by_year")
print(web_sales_by_year_columns)

expected_gold_columns: set[str] = {"d_year", "pedidos", "lineas_de_venta", "beneficio_neto"}
missing_gold_columns: set[str] = expected_gold_columns - set(web_sales_by_year_columns)
assert not missing_gold_columns, (
    "Faltan columnas en iceberg.tcdm.web_sales_by_year: "
    f"{missing_gold_columns}. ¿Se ha ejecutado S7 antes que esta sesión?"
)
print("iceberg.tcdm.web_sales_by_year tiene todas las columnas del contrato de S8.")

create_table_web_sales: str = spark.sql("SHOW CREATE TABLE iceberg.tcdm.web_sales").collect()[0][0]
print(create_table_web_sales)

Generador simulado (Python, se ejecuta varias veces)
        │  escribe ficheros Parquet nuevos
        ▼
/datalake/raw/streaming/web_sales_incremental/batch-<id>.parquet
        │  spark.readStream.format("parquet")
        ▼
Micro-lote de Structured Streaming (Spark)
        │  foreachBatch: vista temporal + MERGE INTO (Spark, catálogo iceberg)
        ▼
iceberg.tcdm.web_sales           (ya creada en S7)
        │  mismo foreachBatch: segundo MERGE INTO (Spark)
        ▼
iceberg.tcdm.web_sales_by_year   (ya creada en S7)

from pathlib import Path
from urllib.request import Request, urlopen

GENERATOR: Path = Path("generate_web_sales_batch.py")
generator_url: str = (
    "https://raw.githubusercontent.com/dsevilla/tcdm-public/26-27/s8/generate_web_sales_batch.py"
)

if GENERATOR.exists():
    print(f"{GENERATOR} ya está en {Path.cwd()}; se conserva.")
else:
    request: Request = Request(generator_url, headers={"User-Agent": "Mozilla/5.0"})
    with urlopen(request, timeout=30) as response:
        GENERATOR.write_bytes(response.read())
    print(f"Descargado {GENERATOR} en {Path.cwd()}")

import json

first_run: subprocess.CompletedProcess[str] = subprocess.run(
    ["python", "generate_web_sales_batch.py", "--rows", "150"],
    capture_output=True,
    text=True,
    check=True,
)
print(first_run.stdout)

first_report: dict[str, object] = json.loads(
    first_run.stdout.strip().splitlines()[-1].split("\t", 1)[1]
)
batch_reports: list[dict[str, object]] = [first_report]
print(first_report)

LANDING_PATH: str = "/datalake/raw/streaming/web_sales_incremental"
print(f"Ruta de aterrizaje: {LANDING_PATH}")

!hdfs dfs -ls {LANDING_PATH}

from pyspark.sql.types import (
    DateType,
    DecimalType,
    IntegerType,
    LongType,
    StructField,
    StructType,
)

web_sales_incremental_schema: StructType = StructType(
    [
        StructField("ws_order_number", LongType(), nullable=False),
        StructField("ws_item_sk", LongType(), nullable=False),
        StructField("ws_bill_customer_sk", LongType(), nullable=True),
        StructField("ws_bill_addr_sk", LongType(), nullable=True),
        StructField("ws_sold_date_sk", LongType(), nullable=True),
        StructField("ws_sold_date", DateType(), nullable=True),
        StructField("ws_quantity", IntegerType(), nullable=True),
        StructField("ws_list_price", DecimalType(7, 2), nullable=True),
        StructField("ws_ext_discount_amt", DecimalType(7, 2), nullable=True),
        StructField("ws_ext_sales_price", DecimalType(7, 2), nullable=True),
        StructField("ws_net_paid", DecimalType(7, 2), nullable=True),
        StructField("ws_net_profit", DecimalType(7, 2), nullable=True),
        StructField("sold_year", IntegerType(), nullable=True),
        StructField("sold_month", IntegerType(), nullable=True),
    ]
)

from pyspark.sql import DataFrame

raw_stream: DataFrame = (
    spark.readStream.format("parquet").schema(web_sales_incremental_schema).load(LANDING_PATH)
)

print("readStream configurado sobre:", LANDING_PATH)

def merge_into_silver_and_gold(micro_batch_df: DataFrame, batch_id: int) -> None:
    '''Aplicar un micro-lote de streaming a la silver y a la gold Iceberg.

    Spark ejecuta aquí los dos MERGE INTO directamente contra el catálogo
    iceberg (Hive Metastore), el mismo que ya verificamos más arriba: no hay
    ningún motor ni tabla intermedios.
    '''

    if micro_batch_df.isEmpty():
        print(f"[foreachBatch] Micro-lote {batch_id}: 0 filas, nada que fusionar.")
        return

    # foreachBatch entrega micro_batch_df con su propia SparkSession "clonada":
    # aunque comparte el mismo catálogo que la sesión global `spark`, es un
    # objeto Python distinto, y una vista temporal es visible sólo dentro de
    # la sesión que la registró. Por eso todo este cuerpo usa
    # `batch_spark`, obtenida del propio DataFrame, en vez de la variable
    # global `spark` capturada por el closure.
    batch_spark: SparkSession = micro_batch_df.sparkSession

    row_count: int = micro_batch_df.count()
    print(f"[foreachBatch] Micro-lote {batch_id}: {row_count} filas recibidas.")

    micro_batch_df.createOrReplaceTempView("micro_batch_src")

    # Años que las claves de este micro-lote tenían ANTES del MERGE: hace
    # falta capturarlos ahora, antes de fusionar, porque una corrección
    # puede cambiar el año de una fila (ver la celda anterior).
    previous_years_df: DataFrame = batch_spark.sql('''
        SELECT DISTINCT target.sold_year AS sold_year
        FROM iceberg.tcdm.web_sales AS target
        JOIN micro_batch_src AS source
          ON target.ws_order_number = source.ws_order_number
             AND target.ws_item_sk = source.ws_item_sk
        ''')
    previous_years: set[int] = {row["sold_year"] for row in previous_years_df.collect()}

    batch_spark.sql('''
        MERGE INTO iceberg.tcdm.web_sales AS target
        USING micro_batch_src AS source
        ON target.ws_order_number = source.ws_order_number
           AND target.ws_item_sk = source.ws_item_sk
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
        ''')

    new_years: set[int] = {
        row["sold_year"] for row in micro_batch_df.select("sold_year").distinct().collect()
    }
    touched_years: list[int] = sorted(previous_years | new_years)
    years_list_sql: str = ", ".join(str(year) for year in touched_years)
    years_values_sql: str = ", ".join(f"({year})" for year in touched_years)
    print(f"[foreachBatch] Micro-lote {batch_id}: años tocados = {touched_years}")

    batch_spark.sql(f'''
        MERGE INTO iceberg.tcdm.web_sales_by_year AS target
        USING (
            SELECT year_list.sold_year AS d_year,
                   COALESCE(agg.pedidos, 0) AS pedidos,
                   COALESCE(agg.lineas_de_venta, 0) AS lineas_de_venta,
                   COALESCE(agg.beneficio_neto, 0) AS beneficio_neto
            FROM (VALUES {years_values_sql}) AS year_list (sold_year)
            LEFT JOIN (
                SELECT sold_year,
                       count(DISTINCT ws_order_number) AS pedidos,
                       count(*) AS lineas_de_venta,
                       sum(ws_net_profit) AS beneficio_neto
                FROM iceberg.tcdm.web_sales
                WHERE sold_year IN ({years_list_sql})
                GROUP BY sold_year
            ) AS agg ON year_list.sold_year = agg.sold_year
        ) AS source
        ON target.d_year = source.d_year
        WHEN MATCHED AND source.lineas_de_venta = 0 THEN DELETE
        WHEN MATCHED THEN UPDATE SET
            pedidos = source.pedidos,
            lineas_de_venta = source.lineas_de_venta,
            beneficio_neto = source.beneficio_neto
        WHEN NOT MATCHED AND source.lineas_de_venta > 0 THEN INSERT (d_year, pedidos, lineas_de_venta, beneficio_neto)
            VALUES (source.d_year, source.pedidos, source.lineas_de_venta, source.beneficio_neto)
        ''')

from pyspark.sql.streaming import StreamingQuery

CHECKPOINT_PATH: str = "/user/luser/s8/checkpoints/web_sales_incremental"

query: StreamingQuery = (
    raw_stream.writeStream.foreachBatch(merge_into_silver_and_gold)
    .trigger(processingTime="5 seconds")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .start()
)

print("Consulta de streaming arrancada. ID:", query.id)

query.processAllAvailable()
print(query.lastProgress)

import time

LOOP_ITERATIONS: int = 3

for iteration in range(1, LOOP_ITERATIONS + 1):
    run: subprocess.CompletedProcess[str] = subprocess.run(
        ["python", "generate_web_sales_batch.py", "--rows", "150", "--update-fraction", "0.4"],
        capture_output=True,
        text=True,
        check=True,
    )
    report: dict[str, object] = json.loads(run.stdout.strip().splitlines()[-1].split("\t", 1)[1])
    batch_reports.append(report)
    print(f"--- Generador, vuelta {iteration}: {report['batch_path']} ({report['rows']} filas) ---")

    query.processAllAvailable()
    print(f"[lastProgress tras la vuelta {iteration}]")
    print(query.lastProgress)

    time.sleep(2)

print(f"Lotes procesados en total: {len(batch_reports)}")

progress_summary: list[dict[str, object]] = [
    {
        "batchId": progress["batchId"],
        "numInputRows": progress["numInputRows"],
        "durationMs_total": progress["durationMs"].get("triggerExecution"),
    }
    for progress in query.recentProgress
]
for entry in progress_summary:
    print(entry)

update_reports_with_keys: list[dict[str, object]] = [
    report for report in batch_reports if report["updated_keys"]
]
assert update_reports_with_keys, "Ningún lote generó correcciones; revisa --update-fraction"

example_update_report: dict[str, object] = update_reports_with_keys[-1]
example_update_order_number, example_update_item_sk = example_update_report["updated_keys"][0]

print(
    f"Clave corregida en el último lote con correcciones: ({example_update_order_number}, {example_update_item_sk})"
)

run_trino(f'''
    SELECT *
    FROM iceberg.tcdm.web_sales
    WHERE ws_order_number = {example_update_order_number}
      AND ws_item_sk = {example_update_item_sk}
    ''')

example_insert_order_number, example_insert_item_sk = first_report["inserted_keys_sample"][0]
print(f"Clave nueva del primer lote: ({example_insert_order_number}, {example_insert_item_sk})")

run_trino(f'''
    SELECT *
    FROM iceberg.tcdm.web_sales
    WHERE ws_order_number = {example_insert_order_number}
      AND ws_item_sk = {example_insert_item_sk}
    ''')

run_trino("SELECT * FROM iceberg.tcdm.web_sales_by_year ORDER BY d_year")

TrinoUserError: Illegal keys in extra_properties: [write.merge.mode]

from pyspark.sql import Row

lab_year_counts: list[Row] = (
    spark.table("iceberg.tcdm.web_sales")
    .groupBy("sold_year")
    .count()
    .withColumnRenamed("count", "lineas")
    .orderBy("lineas", ascending=False)
    .collect()
)
LAB_YEAR: int = int(lab_year_counts[0]["sold_year"])
print(
    f"Año elegido para las tablas de laboratorio: {LAB_YEAR} ({lab_year_counts[0]['lineas']} líneas)"
)

LAB_TABLE_DDL_COLUMNS: str = '''
    ws_order_number BIGINT,
    ws_item_sk BIGINT,
    ws_bill_customer_sk BIGINT,
    ws_bill_addr_sk BIGINT,
    ws_sold_date_sk BIGINT,
    ws_sold_date DATE,
    ws_quantity INT,
    ws_list_price DECIMAL(7, 2),
    ws_ext_discount_amt DECIMAL(7, 2),
    ws_ext_sales_price DECIMAL(7, 2),
    ws_net_paid DECIMAL(7, 2),
    ws_net_profit DECIMAL(7, 2),
    sold_year INT,
    sold_month INT
'''

spark.sql("DROP TABLE IF EXISTS iceberg.tcdm.web_sales_lab_cow")
spark.sql(f'''
    CREATE TABLE iceberg.tcdm.web_sales_lab_cow ({LAB_TABLE_DDL_COLUMNS})
    USING iceberg
    PARTITIONED BY (months(ws_sold_date))
    TBLPROPERTIES (
        'format-version' = '2',
        'write.merge.mode' = 'copy-on-write',
        'write.update.mode' = 'copy-on-write',
        'write.delete.mode' = 'copy-on-write'
    )
''')

spark.sql("DROP TABLE IF EXISTS iceberg.tcdm.web_sales_lab_mor")
spark.sql(f'''
    CREATE TABLE iceberg.tcdm.web_sales_lab_mor ({LAB_TABLE_DDL_COLUMNS})
    USING iceberg
    PARTITIONED BY (months(ws_sold_date))
    TBLPROPERTIES (
        'format-version' = '2',
        'write.merge.mode' = 'merge-on-read',
        'write.update.mode' = 'merge-on-read',
        'write.delete.mode' = 'merge-on-read'
    )
''')
print(
    "Tablas de laboratorio creadas: web_sales_lab_cow (copy-on-write), web_sales_lab_mor (merge-on-read)."
)

from pyspark.sql.functions import col as spark_col

lab_source_df: DataFrame = (
    spark.table("iceberg.tcdm.web_sales")
    .filter(spark_col("sold_year") == LAB_YEAR)
    .select([field.name for field in web_sales_incremental_schema.fields])
)

lab_source_df.write.mode("append").saveAsTable("iceberg.tcdm.web_sales_lab_cow")
lab_source_df.write.mode("append").saveAsTable("iceberg.tcdm.web_sales_lab_mor")

lab_source_rows: int = lab_source_df.count()
cow_rows: int = spark.table("iceberg.tcdm.web_sales_lab_cow").count()
mor_rows: int = spark.table("iceberg.tcdm.web_sales_lab_mor").count()
print(f"Filas cargadas: cow={cow_rows}, mor={mor_rows}")
assert cow_rows == mor_rows == lab_source_rows

import random
from datetime import date

from pyspark.sql import Row

lab_sample_keys: list[Row] = (
    spark.table("iceberg.tcdm.web_sales_lab_cow")
    .select("ws_order_number", "ws_item_sk")
    .limit(3)
    .collect()
)

lab_rng: random.Random = random.Random(2027)
lab_rows: list[dict[str, object]] = []

for key_row in lab_sample_keys:
    lab_rows.append(
        {
            "ws_order_number": key_row["ws_order_number"],
            "ws_item_sk": key_row["ws_item_sk"],
            "ws_bill_customer_sk": lab_rng.randint(1, 100_000),
            "ws_bill_addr_sk": lab_rng.randint(1, 50_000),
            "ws_sold_date_sk": None,
            "ws_sold_date": date(LAB_YEAR, 6, 15),
            "ws_quantity": lab_rng.randint(1, 20),
            "ws_list_price": None,
            "ws_ext_discount_amt": None,
            "ws_ext_sales_price": None,
            "ws_net_paid": None,
            "ws_net_profit": None,
            "sold_year": LAB_YEAR,
            "sold_month": 6,
        }
    )

for offset in range(2):
    lab_rows.append(
        {
            "ws_order_number": 950_000_000 + offset,
            "ws_item_sk": lab_rng.randint(1, 18_000),
            "ws_bill_customer_sk": lab_rng.randint(1, 100_000),
            "ws_bill_addr_sk": lab_rng.randint(1, 50_000),
            "ws_sold_date_sk": None,
            "ws_sold_date": date(LAB_YEAR, 6, 15),
            "ws_quantity": lab_rng.randint(1, 20),
            "ws_list_price": None,
            "ws_ext_discount_amt": None,
            "ws_ext_sales_price": None,
            "ws_net_paid": None,
            "ws_net_profit": None,
            "sold_year": LAB_YEAR,
            "sold_month": 6,
        }
    )

lab_merge_batch: DataFrame = spark.createDataFrame(lab_rows, schema=web_sales_incremental_schema)
lab_merge_batch.createOrReplaceTempView("lab_merge_batch_view")
lab_merge_batch.show(truncate=False)

from pyspark.sql import Row


def table_location(table_name: str) -> str:
    '''Ubicación física real de una tabla del catálogo iceberg (Hive Metastore).

    A diferencia de un catálogo hadoop, donde la ruta se puede calcular sin
    preguntarle a nadie, aquí el Hive Metastore puede añadir un sufijo único
    a la ubicación de cada tabla (como ya viste en S7 con
    "SHOW CREATE TABLE"): hay que preguntárselo al catálogo con
    "DESCRIBE TABLE EXTENDED", la versión Spark del mismo tipo de consulta.
    '''
    extended_rows: list[Row] = spark.sql(f"DESCRIBE TABLE EXTENDED {table_name}").collect()
    location_row: Row = next(row for row in extended_rows if row["col_name"] == "Location")
    return location_row["data_type"]


cow_location: str = table_location("iceberg.tcdm.web_sales_lab_cow")
mor_location: str = table_location("iceberg.tcdm.web_sales_lab_mor")
print(f"cow: {cow_location}")
print(f"mor: {mor_location}")

!hdfs dfs -ls -R {cow_location}/data

!hdfs dfs -ls -R {mor_location}/data

spark.sql('''
    MERGE INTO iceberg.tcdm.web_sales_lab_cow AS target
    USING lab_merge_batch_view AS source
    ON target.ws_order_number = source.ws_order_number
       AND target.ws_item_sk = source.ws_item_sk
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
''')

spark.sql('''
    MERGE INTO iceberg.tcdm.web_sales_lab_mor AS target
    USING lab_merge_batch_view AS source
    ON target.ws_order_number = source.ws_order_number
       AND target.ws_item_sk = source.ws_item_sk
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
''')
print("MERGE aplicado a las dos tablas de laboratorio.")

!hdfs dfs -ls -R {cow_location}/data

!hdfs dfs -ls -R {mor_location}/data

print("query.status:", query.status)
print("query.lastProgress:", query.lastProgress)
print(f"Micro-lotes acumulados en recentProgress: {len(query.recentProgress)}")

query.stop()
print(f"Consulta detenida. isActive = {query.isActive}")

query_restarted: StreamingQuery = (
    raw_stream.writeStream.foreachBatch(merge_into_silver_and_gold)
    .trigger(processingTime="5 seconds")
    .option("checkpointLocation", CHECKPOINT_PATH)
    .start()
)

query_restarted.processAllAvailable()
restarted_progress: dict[str, object] | None = query_restarted.lastProgress
print(restarted_progress)

query_restarted.stop()
print(f"Segunda consulta detenida. isActive = {query_restarted.isActive}")

batch_recomputed_gold: pd.DataFrame = run_trino('''
    SELECT sold_year AS d_year,
           count(DISTINCT ws_order_number) AS pedidos,
           count(*) AS lineas_de_venta,
           sum(ws_net_profit) AS beneficio_neto
    FROM iceberg.tcdm.web_sales
    GROUP BY sold_year
    ORDER BY d_year
    ''')

incremental_gold: pd.DataFrame = run_trino(
    "SELECT * FROM iceberg.tcdm.web_sales_by_year ORDER BY d_year"
)

comparison: pd.DataFrame = batch_recomputed_gold.merge(
    incremental_gold, on="d_year", suffixes=("_batch", "_incremental")
)
comparison["beneficio_neto_batch"] = comparison["beneficio_neto_batch"].astype(float)
comparison["beneficio_neto_incremental"] = comparison["beneficio_neto_incremental"].astype(float)
comparison["diferencia_beneficio"] = (
    comparison["beneficio_neto_batch"] - comparison["beneficio_neto_incremental"]
).abs()

assert (
    len(comparison) == len(batch_recomputed_gold) == len(incremental_gold)
), "Faltan años en una de las dos versiones"
assert (comparison["pedidos_batch"] == comparison["pedidos_incremental"]).all()
assert (comparison["lineas_de_venta_batch"] == comparison["lineas_de_venta_incremental"]).all()
assert (comparison["diferencia_beneficio"] < 0.01).all()

print("El agregado incremental coincide, año a año, con el recálculo completo por lotes.")
comparison

key_uniqueness: pd.DataFrame = run_trino('''
    WITH totals AS (
        SELECT count(*) AS filas FROM iceberg.tcdm.web_sales
    ),
    distinct_keys AS (
        SELECT count(*) AS claves FROM (
            SELECT DISTINCT ws_order_number, ws_item_sk FROM iceberg.tcdm.web_sales
        )
    )
    SELECT filas, claves FROM totals, distinct_keys
''')
assert int(key_uniqueness.loc[0, "filas"]) == int(key_uniqueness.loc[0, "claves"]), (
    "La clave compuesta (ws_order_number, ws_item_sk) debe seguir siendo única "
    "en iceberg.tcdm.web_sales tras los MERGE de esta sesión"
)
key_uniqueness

snapshots_web_sales_tras_streaming: pd.DataFrame = run_trino('''
    SELECT snapshot_id, committed_at, operation
    FROM iceberg.tcdm."web_sales$snapshots"
    ORDER BY committed_at
''')
snapshots_web_sales_tras_streaming

Concepto	Significado en esta sesión	Lo que no es
Generador simulado	Proceso simple que escribe nuevos ficheros Parquet en una ruta raw de aterrizaje	Un sistema de colas o mensajería
Fuente de streaming	`spark.readStream` vigilando esa ruta y detectando ficheros nuevos	Una conexión a un broker
Micro-lote (micro-batch)	Conjunto de filas nuevas que Structured Streaming procesa en una ejecución del disparador	Un evento individual procesado uno a uno
Disparador (trigger)	Cada cuánto tiempo, o con qué condición, se procesa un micro-lote	Un `cron` del sistema operativo
Sink con `foreachBatch`	Función que recibe el DataFrame de cada micro-lote y decide qué hacer con él	Una escritura directa fila a fila
Merge (Iceberg)	Operación `MERGE INTO` que inserta, actualiza o ignora filas según coincidan con la tabla destino	Sobrescribir toda la tabla

Situación	Orden	Consecuencia
Eliminar las tablas de laboratorio copy-on-write/merge-on-read	`spark.sql("DROP TABLE iceberg.tcdm.web_sales_lab_cow")` y lo mismo para `web_sales_lab_mor`	Borra las dos tablas del catálogo `iceberg` (metadatos en el Hive Metastore y ficheros bajo `/warehouse`). No afecta a `iceberg.tcdm.web_sales` ni a `iceberg.tcdm.web_sales_by_year`: son tablas distintas.
Vaciar la ruta de aterrizaje del streaming	`hdfs dfs -rm -r /datalake/raw/streaming/web_sales_incremental`	Borra todos los lotes Parquet generados por `generate_web_sales_batch.py` en esta sesión. No afecta a `/datalake/raw/tpcds`.
Forzar que la consulta reprocese todo desde el principio	`hdfs dfs -rm -r /user/luser/s8/checkpoints/web_sales_incremental`	Borra el checkpoint: la próxima vez que se arranque la consulta sobre la misma ruta de aterrizaje, volverá a fusionar todos los ficheros que sigan ahí. Útil para repetir el ejercicio desde cero, nunca como efecto colateral de otra limpieza.
Volver al estado en que S7 dejó las tablas	Con la consulta detenida: `spark.sql("DROP TABLE iceberg.tcdm.web_sales")` y `spark.sql("DROP TABLE iceberg.tcdm.web_sales_by_year")`; borrar la ruta de aterrizaje y el checkpoint (las dos filas anteriores); y volver a ejecutar en S7 las celdas de «Crear la tabla Iceberg»	Deshace todo lo que esta sesión ha fusionado: S7 recrea las dos tablas desde las tablas Hive de S6, que no han cambiado. Si no se borran el aterrizaje y el checkpoint, la siguiente ejecución de esta sesión volvería a aplicar los lotes antiguos.
Parar temporalmente el warehouse	`docker compose -f entorno/compose-warehouse-hdfs.yml stop`	Conserva PostgreSQL, Hive Metastore y Trino (y su contenido); se reanuda con `start`. No afecta al clúster Hadoop ni a `iceberg.tcdm.web_sales`.

Sesión 8 — Ingesta continua con Spark Structured Streaming¶

Diapositivas de la sesión¶

Sesión 8: ingesta continua con Spark Structured Streaming

Del lote único a las tandas

Recorrido del curso¶

Qué significa "streaming" aquí, y qué no¶

Objetivos de la sesión¶

Cómo se evalúa esta sesión

Antes de empezar¶

Instalar las dependencias de este notebook¶

Configurar el catálogo Iceberg de Spark

Cómo configuramos el catálogo Iceberg de Spark¶

Quién escribe y quién comprueba

Construir la SparkSession¶

Abrir la conexión con Trino¶

Verificar que S7 ya creó las tablas Iceberg¶

Arquitectura y generador simulado

Arquitectura de la sesión¶

El generador simulado¶

Una fuente que vigila un directorio

La consulta de streaming¶

Qué hace foreachBatch con cada micro-lote

foreachBatch: vista temporal + dos MERGE INTO¶

Arrancar la consulta¶

Un micro-lote, dos fusiones

Primeros micro-lotes¶

MERGE INTO hacia silver, explicado¶

De silver a gold, de forma incremental¶

Agregar no es recalcular todo

Dos formas de escribir lo mismo

Copy-on-write frente a merge-on-read¶

El coste se paga al escribir o al leer

Observar el progreso y detener la consulta¶

Reiniciar desde el mismo checkpoint¶

La prueba de que el streaming no cambia el resultado

Comprobar la coherencia con el recorrido por lotes¶

Mismo SQL, más a menudo y sobre menos datos

Qué no se ha visto todavía¶

Detener y limpiar¶

Preguntas para interpretar la experiencia¶

Evidencias de la sesión¶

Qué mostrar en el ordenador durante la reunión¶

Memoria escrita (una o dos páginas)¶

Qué es importante de cara al examen final¶

Sesión 8

Siguiente paso¶

Qué hace `foreachBatch` con cada micro-lote

`foreachBatch`: vista temporal + dos `MERGE INTO`¶

`MERGE INTO` hacia silver, explicado¶