S1: HDFS, YARN y MapReduce
  → S2: Parquet en /datalake/raw/tpcds, sin catálogo permanente
  → S3: Parquet sobre un almacenamiento S3-compatible
  → S4: Spark procesa los Parquet y utiliza YARN

%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")

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

!hdfs dfs -ls -d /datalake/raw/tpcds

%%writefile requirements.txt
# Dependencias del kernel interactivo de la sesión 4 (Spark en modo local).
# La serie de PySpark aceptada por el curso es >4,<4.2.
# El envío a YARN de la sección "Driver, ejecutores y YARN" usa este mismo
# PySpark (spark-submit de /opt/tcdm/venv).
pyspark>4,<4.2

%pip install -q -r requirements.txt

import sys

import pyspark

print(f"PySpark {pyspark.__version__} con Python {sys.version.split()[0]}.")

from pyspark import SparkContext
from pyspark.sql import SparkSession

spark: SparkSession = (
    SparkSession.builder.appName("TCDM_S4_Local")
    .master("local[2]")
    .config("spark.driver.memory", "1g")
    .config("spark.sql.shuffle.partitions", "8")
    .getOrCreate()
)
sc: SparkContext = spark.sparkContext

print(f"Spark version: {spark.version} | Master: {sc.master} | App: {sc.appName}")

from pyspark.sql.dataframe import DataFrame

date_dim_path: str = "/datalake/raw/tpcds/date_dim"
date_dim: DataFrame = spark.read.parquet(date_dim_path)

print(f"Ruta física leída: {date_dim_path}")
date_dim.printSchema()

date_dim.show(5)

n_date_dim: int = date_dim.count()
print(f"Filas en date_dim: {n_date_dim}")
print(f"Particiones de ejecución: {date_dim.rdd.getNumPartitions()}")

assert n_date_dim == 73_049

web_sales_path: str = "/datalake/raw/tpcds/web_sales"
web_sales: DataFrame = spark.read.parquet(web_sales_path)

print(f"Ruta física leída: {web_sales_path}")
web_sales.printSchema()

web_sales.show(5)

n_web_sales: int = web_sales.count()
print(f"Filas en web_sales: {n_web_sales}")
print(f"Particiones de ejecución: {web_sales.rdd.getNumPartitions()}")

assert n_web_sales == 719_384

from pyspark.sql.functions import col

web_sales_profitable: DataFrame = (
    web_sales.select(
        "ws_order_number",
        "ws_bill_customer_sk",
        "ws_net_profit",
        "ws_ext_sales_price",
    )
    .filter(col("ws_net_profit") > 0)
    .withColumn("margen_sobre_venta", col("ws_net_profit") / col("ws_ext_sales_price"))
)

print(f"Tipo del objeto: {type(web_sales_profitable)}")

web_sales_profitable.show(5)

narrow_query: DataFrame = web_sales.select("ws_order_number", "ws_net_profit").filter(
    col("ws_net_profit") > 0
)
wide_query: DataFrame = web_sales.select("*").filter(col("ws_net_profit") > 0)

print("Plan de la consulta con columnas seleccionadas:")
narrow_query.explain("formatted")

print("Plan de la consulta con select(\"*\"):")
wide_query.explain("formatted")

lineas_con_beneficio: DataFrame = (
    web_sales.filter((col("ws_net_profit") > 0) & col("ws_bill_customer_sk").isNotNull())
    .groupBy("ws_bill_customer_sk")
    .count()
    .orderBy(col("count").desc())
)
lineas_con_beneficio.show(10)

item: DataFrame = spark.read.parquet("/datalake/raw/tpcds/item")

web_sales_dated_categorized: DataFrame = (
    web_sales.join(date_dim, web_sales.ws_sold_date_sk == date_dim.d_date_sk)
    .join(item, web_sales.ws_item_sk == item.i_item_sk)
    .select("d_year", "i_category", "ws_net_profit")
)

n_item: int = item.count()
print(f"Filas en item: {n_item}")
assert n_item == 18_000

from pyspark.sql.functions import sum as spark_sum

rollup_year_category: DataFrame = (
    web_sales_dated_categorized.rollup("d_year", "i_category")
    .agg(spark_sum("ws_net_profit").alias("beneficio_neto"))
    .orderBy("d_year", "i_category")
)
rollup_year_category.show(100)

cube_year_category: DataFrame = (
    web_sales_dated_categorized.cube("d_year", "i_category")
    .agg(spark_sum("ws_net_profit").alias("beneficio_neto"))
    .orderBy("d_year", "i_category")
)
cube_year_category.show(100)

import pyspark.sql.functions as F

customer: DataFrame = spark.read.parquet("/datalake/raw/tpcds/customer")
customer_address: DataFrame = spark.read.parquet("/datalake/raw/tpcds/customer_address")

n_customer: int = customer.count()
n_customer_address: int = customer_address.count()
print(f"Filas en customer: {n_customer}")
print(f"Filas en customer_address: {n_customer_address}")
assert n_customer == 100_000
assert n_customer_address == 50_000

# Mismo filtro que usó S2 para reconocer las direcciones ya cruzadas con el
# INE: país fijado a España, municipio y provincia no nulos, y código de
# provincia de dos dígitos.
localized_addresses: DataFrame = customer_address.filter(
    (F.trim(F.col("ca_country")) == "España")
    & F.col("ca_city").isNotNull()
    & F.col("ca_county").isNotNull()
    & F.trim(F.col("ca_state")).rlike(r"^[0-9]{2}$")
)
assert localized_addresses.count() == n_customer_address

web_sales_slim: DataFrame = web_sales.select(
    "ws_order_number", "ws_bill_customer_sk", "ws_bill_addr_sk", "ws_net_profit"
)

from pyspark.sql.column import Column

join_expr: Column = web_sales_slim.ws_bill_addr_sk == customer_address.ca_address_sk

inner: DataFrame = web_sales_slim.join(customer_address, join_expr, "inner")
print(f"INNER JOIN: {inner.count()} filas.")

left_outer: DataFrame = web_sales_slim.join(customer_address, join_expr, "left_outer")
print(f"LEFT OUTER JOIN: {left_outer.count()} filas (incluye líneas sin domicilio conocido).")

right_outer: DataFrame = web_sales_slim.join(customer_address, join_expr, "right_outer")
print(f"RIGHT OUTER JOIN: {right_outer.count()} filas (incluye domicilios sin ventas).")

left_semi: DataFrame = web_sales_slim.join(localized_addresses, join_expr, "left_semi")
print(
    f"LEFT SEMI JOIN: {left_semi.count()} filas de web_sales cuyo domicilio existe"
    " y está localizado (municipio del INE) en customer_address."
)

left_anti: DataFrame = web_sales_slim.join(customer_address, join_expr, "left_anti")
print(
    f"LEFT ANTI JOIN: {left_anti.count()} filas de web_sales cuyo domicilio NO existe en customer_address."
)

# Producto cartesiano: cada fila de web_sales_slim con TODAS las filas de
# customer_address. NO debe ejecutarse: el resultado tendría
# aproximadamente 719.384 x 50.000 filas.
# cross = web_sales_slim.crossJoin(customer_address)

from pyspark.sql.functions import broadcast

sales_with_state: DataFrame = web_sales_slim.join(broadcast(customer_address), join_expr, "inner")

print("Plan físico del broadcast join:")
sales_with_state.explain("formatted")

web_sales.select(
    F.avg("ws_ext_sales_price").alias("precio_medio"),
    F.max("ws_ext_sales_price").alias("precio_maximo"),
    F.min("ws_ext_sales_price").alias("precio_minimo"),
    F.count("ws_ext_sales_price").alias("lineas_con_precio"),
).show()

web_sales.select("ws_ext_sales_price", "ws_net_profit").describe().show()

orders_base: DataFrame = web_sales.groupBy("ws_bill_customer_sk", "ws_order_number").agg(
    F.sum("ws_net_paid").alias("importe_pedido"),
    F.first("ws_sold_date_sk").alias("ws_sold_date_sk"),
)
print(f"Pedidos distintos agregados: {orders_base.count()}")

customer_order_history: DataFrame = orders_base.groupBy("ws_bill_customer_sk").agg(
    F.collect_list(
        F.struct(
            col("ws_order_number"),
            col("importe_pedido"),
            col("ws_sold_date_sk"),
        )
    ).alias("historial_pedidos")
)
customer_order_history.printSchema()
customer_order_history.show(5, truncate=False)

customer_order_history.select(
    "ws_bill_customer_sk",
    F.col("historial_pedidos").getItem(0).alias("primer_pedido_del_array"),
    F.size("historial_pedidos").alias("numero_de_pedidos"),
).orderBy(F.col("numero_de_pedidos").desc()).show(10, truncate=False)

clientes_con_muchos_pedidos: DataFrame = customer_order_history.filter(
    F.size("historial_pedidos") > 5
)
print(f"Clientes con más de 5 pedidos: {clientes_con_muchos_pedidos.count()}")

exploded_orders: DataFrame = customer_order_history.select(
    "ws_bill_customer_sk", F.explode("historial_pedidos").alias("pedido")
)
exploded_orders.select(
    "ws_bill_customer_sk",
    col("pedido.ws_order_number").alias("pedido_numero"),
    col("pedido.importe_pedido").alias("importe"),
).orderBy("ws_bill_customer_sk", "pedido_numero").show(10)

from decimal import Decimal

from pyspark.sql.types import StringType


@F.udf(returnType=StringType())
def clasificar_beneficio(net_profit: Decimal | None) -> str:
    # Clasifica una línea de venta según su beneficio neto (ws_net_profit).
    if net_profit is None:
        return "desconocido"
    if net_profit < 0:
        return "perdida"
    if net_profit < 50:
        return "margen_bajo"
    return "margen_alto"


web_sales_classified: DataFrame = web_sales.withColumn(
    "clasificacion_beneficio", clasificar_beneficio(col("ws_net_profit"))
)
web_sales_classified.groupBy("clasificacion_beneficio").count().orderBy(
    F.col("count").desc()
).show()

web_sales_classified_builtin: DataFrame = web_sales.withColumn(
    "clasificacion_beneficio",
    F.when(col("ws_net_profit").isNull(), "desconocido")
    .when(col("ws_net_profit") < 0, "perdida")
    .when(col("ws_net_profit") < 50, "margen_bajo")
    .otherwise("margen_alto"),
)
web_sales_classified_builtin.groupBy("clasificacion_beneficio").count().orderBy(
    F.col("count").desc()
).show()

from pyspark.sql.window import Window, WindowSpec

orders_dated: DataFrame = orders_base.join(
    date_dim, orders_base.ws_sold_date_sk == date_dim.d_date_sk
).select(
    col("ws_bill_customer_sk").alias("cliente"),
    col("ws_order_number").alias("pedido"),
    col("d_date").alias("fecha_pedido"),
    col("importe_pedido"),
)
orders_dated.orderBy("cliente", "fecha_pedido").show(10)

from pyspark.sql.column import Column

ventana_por_cliente: WindowSpec = Window.partitionBy("cliente")
importe_maximo_cliente: Column = F.max("importe_pedido").over(ventana_por_cliente)

orders_dated.select(
    "cliente",
    "pedido",
    "importe_pedido",
    importe_maximo_cliente.alias("pedido_mayor_del_cliente"),
).withColumn(
    "es_su_pedido_mayor", col("importe_pedido") == col("pedido_mayor_del_cliente")
).orderBy(
    "cliente", "fecha_pedido"
).show(
    15
)

ventana_ordenada: WindowSpec = Window.partitionBy("cliente").orderBy("fecha_pedido")

fecha_pedido_anterior: Column = F.lag("fecha_pedido", 1).over(ventana_ordenada)
fecha_pedido_siguiente: Column = F.lead("fecha_pedido", 1).over(ventana_ordenada)

orders_dated.select(
    "cliente",
    "pedido",
    "fecha_pedido",
    F.datediff("fecha_pedido", fecha_pedido_anterior).alias("dias_desde_pedido_anterior"),
    F.datediff(fecha_pedido_siguiente, "fecha_pedido").alias("dias_hasta_pedido_siguiente"),
).orderBy("cliente", "fecha_pedido").show(15)

ventana_movil: WindowSpec = ventana_ordenada.rowsBetween(-2, 0)
media_movil_3_pedidos: Column = F.avg("importe_pedido").over(ventana_movil)

orders_dated.select(
    "cliente",
    "pedido",
    "fecha_pedido",
    "importe_pedido",
    media_movil_3_pedidos.alias("media_movil_3_pedidos"),
).orderBy("cliente", "fecha_pedido").show(15)

ventana_acumulada: WindowSpec = ventana_ordenada.rowsBetween(
    Window.unboundedPreceding, Window.currentRow
)
gasto_acumulado: Column = F.sum("importe_pedido").over(ventana_acumulada)

orders_dated.select(
    "cliente",
    "pedido",
    "fecha_pedido",
    "importe_pedido",
    gasto_acumulado.alias("gasto_acumulado_cliente"),
).orderBy("cliente", "fecha_pedido").show(15)

web_sales.createOrReplaceTempView("web_sales")
spark.sql("SELECT ... FROM web_sales")

web_sales.createOrReplaceTempView("web_sales")
customer.createOrReplaceTempView("customer")
customer_address.createOrReplaceTempView("customer_address")
date_dim.createOrReplaceTempView("date_dim")

time_dim: DataFrame = spark.read.parquet("/datalake/raw/tpcds/time_dim")
n_time_dim: int = time_dim.count()
print(f"Filas en time_dim: {n_time_dim}")
assert n_time_dim == 86_400
time_dim.createOrReplaceTempView("time_dim")

spark.sql("""
    SELECT
        c.c_customer_id,
        count(DISTINCT s.ws_order_number) AS pedidos,
        count(*) AS lineas_de_venta
    FROM web_sales AS s
    JOIN customer AS c
        ON s.ws_bill_customer_sk = c.c_customer_sk
    GROUP BY c.c_customer_id
    ORDER BY pedidos DESC, c.c_customer_id
    LIMIT 10
""").show()

spark.sql("""
    SELECT
        a.ca_state AS codigo_provincia,
        a.ca_county AS provincia,
        count(DISTINCT s.ws_order_number) AS pedidos,
        sum(s.ws_ext_sales_price) AS ventas
    FROM web_sales AS s
    JOIN customer_address AS a
        ON s.ws_bill_addr_sk = a.ca_address_sk
    GROUP BY a.ca_state, a.ca_county
    ORDER BY pedidos DESC, a.ca_state
    LIMIT 10
""").show()

spark.sql("""
    SELECT
        t.t_hour,
        count(DISTINCT s.ws_order_number) AS pedidos,
        sum(s.ws_ext_sales_price) AS ventas
    FROM web_sales AS s
    JOIN time_dim AS t
        ON s.ws_sold_time_sk = t.t_time_sk
    GROUP BY t.t_hour
    ORDER BY t.t_hour
""").show(24)

from pyspark.sql import Row

sales_by_year_local: DataFrame = spark.sql("""
    SELECT
        d.d_year,
        sum(s.ws_net_profit) AS beneficio_neto_total,
        count(DISTINCT s.ws_order_number) AS pedidos_distintos,
        count(*) AS lineas_de_venta
    FROM web_sales AS s
    JOIN date_dim AS d
        ON s.ws_sold_date_sk = d.d_date_sk
    GROUP BY d.d_year
    ORDER BY d.d_year
""")
sales_by_year_local.show()

# Se guarda como filas de Python: la SparkSession local se cierra antes de
# lanzar el trabajo sobre YARN y el DataFrame deja de poder usarse.
sales_by_year_local_rows: list[Row] = sales_by_year_local.collect()

!hdfs dfs -mkdir -p /user/luser/s4

sales_by_year_category: DataFrame = web_sales_dated_categorized.groupBy("d_year", "i_category").agg(
    spark_sum("ws_net_profit").alias("beneficio_neto")
)
print(f"Filas a escribir: {sales_by_year_category.count()}")

many_files_path: str = "/user/luser/s4/sales-by-year-category-muchos-ficheros"
sales_by_year_category.repartition(50).write.mode("overwrite").parquet(many_files_path)

!hdfs dfs -ls {many_files_path}

few_files_path: str = "/user/luser/s4/sales-by-year-category-pocos-ficheros"
sales_by_year_category.coalesce(4).write.mode("overwrite").parquet(few_files_path)

!hdfs dfs -ls {few_files_path}

print(f"web_sales_slim particiones (antes): {web_sales_slim.rdd.getNumPartitions()}")
print(f"customer_address particiones (antes): {customer_address.rdd.getNumPartitions()}")

web_sales_by_addr: DataFrame = web_sales_slim.repartition(8, "ws_bill_addr_sk")
customer_address_by_key: DataFrame = customer_address.repartition(8, "ca_address_sk")

print(f"web_sales_by_addr particiones (después): {web_sales_by_addr.rdd.getNumPartitions()}")
print(
    f"customer_address_by_key particiones (después): {customer_address_by_key.rdd.getNumPartitions()}"
)

join_expr_repartitioned: Column = (
    web_sales_by_addr.ws_bill_addr_sk == customer_address_by_key.ca_address_sk
)
sales_with_state_repartitioned: DataFrame = web_sales_by_addr.join(
    customer_address_by_key, join_expr_repartitioned, "inner"
)

print("Plan físico tras reparticionar por la clave de unión:")
sales_with_state_repartitioned.explain("formatted")

spark.conf.set("spark.sql.adaptive.enabled", "false")

store_sales: DataFrame = spark.read.parquet("/datalake/raw/tpcds/store_sales")
n_store_sales: int = store_sales.count()
print(f"Filas en store_sales: {n_store_sales}")
assert n_store_sales == 2_880_404

print(f"spark.sql.shuffle.partitions actual: {spark.conf.get('spark.sql.shuffle.partitions')}")

store_sales_by_year: DataFrame = (
    store_sales.join(date_dim, store_sales.ss_sold_date_sk == date_dim.d_date_sk)
    .groupBy("d_year")
    .agg(spark_sum("ss_ext_sales_price").alias("ventas_tienda"))
)
print(
    f"Particiones tras el groupBy (shuffle.partitions=8): {store_sales_by_year.rdd.getNumPartitions()}"
)

spark.conf.set("spark.sql.shuffle.partitions", 3)

store_sales_by_year_3: DataFrame = (
    store_sales.join(date_dim, store_sales.ss_sold_date_sk == date_dim.d_date_sk)
    .groupBy("d_year")
    .agg(spark_sum("ss_ext_sales_price").alias("ventas_tienda"))
)
print(
    f"Particiones tras el groupBy (shuffle.partitions=3): {store_sales_by_year_3.rdd.getNumPartitions()}"
)

# Volvemos a los valores usados en el resto del notebook
spark.conf.set("spark.sql.shuffle.partitions", 8)
spark.conf.set("spark.sql.adaptive.enabled", "true")

from pyspark.sql.types import Row

year_bounds: Row | None = date_dim.agg(
    F.min("d_year").alias("min_year"), F.max("d_year").alias("max_year")
).first()
sample_year: int = (year_bounds["min_year"] + year_bounds["max_year"]) // 2
print(
    f"Años en date_dim: {year_bounds['min_year']}-{year_bounds['max_year']}. Usamos {sample_year}."
)

spark.conf.set("spark.sql.shuffle.partitions", 200)
spark.conf.set("spark.sql.adaptive.enabled", "false")

sin_aqe: DataFrame = (
    store_sales.join(date_dim, store_sales.ss_sold_date_sk == date_dim.d_date_sk)
    .filter(col("d_year") == sample_year)
    .groupBy("d_year")
    .agg(spark_sum("ss_ext_sales_price").alias("ventas_tienda"))
)
sin_aqe.collect()
print(f"Particiones sin AQE (shuffle.partitions=200): {sin_aqe.rdd.getNumPartitions()}")

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

con_aqe: DataFrame = (
    store_sales.join(date_dim, store_sales.ss_sold_date_sk == date_dim.d_date_sk)
    .filter(col("d_year") == sample_year)
    .groupBy("d_year")
    .agg(spark_sum("ss_ext_sales_price").alias("ventas_tienda"))
)
con_aqe.collect()
print(f"Particiones con AQE (coalescePartitions habilitado): {con_aqe.rdd.getNumPartitions()}")

# Volvemos al valor usado en el resto del notebook
spark.conf.set("spark.sql.shuffle.partitions", 8)

print(f"¿Está cacheado antes de cache()?: {sales_with_state.is_cached}")

sales_with_state.cache()
sales_with_state.count()  # acción que materializa la cache

print(f"¿Está cacheado después de cache()?: {sales_with_state.is_cached}")
print(f"Nivel de almacenamiento: {sales_with_state.storageLevel}")

sales_with_state.groupBy("ca_state", "ca_county").count().orderBy(F.col("count").desc()).show(5)
sales_with_state.agg(F.avg("ws_net_profit")).show()

sales_with_state_profitable: DataFrame = sales_with_state.filter(col("ws_net_profit") > 0)
print(f"¿Está cacheado el DataFrame derivado?: {sales_with_state_profitable.is_cached}")

from pyspark import StorageLevel

sales_with_state.unpersist()
print(f"¿Está cacheado tras unpersist()?: {sales_with_state.is_cached}")

sales_with_state.persist(StorageLevel.MEMORY_AND_DISK)
sales_with_state.count()
print(f"Nuevo nivel de almacenamiento: {sales_with_state.storageLevel}")
sales_with_state.unpersist()

spark.stop()

from pathlib import Path

JOB_SCRIPT: str = r'''
# Trabajo independiente para YARN, lanzado con spark-submit --master yarn.
#
# No es la SparkSession del notebook: Spark no permite cambiar el master de
# una sesión ya construida, así que esta aplicación crea la suya propia.
import socket

from pyspark.sql import SparkSession
from pyspark.sql.functions import count, countDistinct
from pyspark.sql.functions import sum as spark_sum
from pyspark.sql.types import IntegerType, LongType, StringType, StructField, StructType

BUSINESS_OUTPUT = "/user/luser/s4/yarn-job/business-result"
EXECUTOR_REPORT_OUTPUT = "/user/luser/s4/yarn-job/executor-report"

spark = SparkSession.builder.appName("TCDM_S4_YARN_business").master("yarn").getOrCreate()

date_dim = spark.read.parquet("/datalake/raw/tpcds/date_dim")
web_sales = spark.read.parquet("/datalake/raw/tpcds/web_sales")

sales_by_year = (
    web_sales.join(date_dim, web_sales.ws_sold_date_sk == date_dim.d_date_sk)
    .groupBy("d_year")
    .agg(
        spark_sum("ws_net_profit").alias("beneficio_neto_total"),
        countDistinct("ws_order_number").alias("pedidos_distintos"),
        count("*").alias("lineas_de_venta"),
    )
    .orderBy("d_year")
)
sales_by_year.coalesce(1).write.mode("overwrite").csv(BUSINESS_OUTPUT, header=True)

# Registra el host que ha ejecutado cada partición.
def describe_partition(partition_id, rows):
    rows = list(rows)
    yield (partition_id, socket.gethostname(), len(rows))


reports = (
    web_sales.select("ws_order_number").rdd.mapPartitionsWithIndex(describe_partition).collect()
)

report_schema = StructType(
    [
        StructField("partition", IntegerType(), False),
        StructField("executor_host", StringType(), False),
        StructField("records", LongType(), False),
    ]
)
(
    spark.createDataFrame(reports, report_schema)
    .orderBy("partition")
    .coalesce(1)
    .write.mode("overwrite")
    .csv(EXECUTOR_REPORT_OUTPUT, header=True)
)

distinct_hosts = {report[1] for report in reports}
print(f"Hosts remotos que procesaron alguna partición: {sorted(distinct_hosts)}")
spark.stop()
'''

Path("/tmp/s4_yarn_job.py").write_text(JOB_SCRIPT)
print("Escrito /tmp/s4_yarn_job.py")

import subprocess

result: subprocess.CompletedProcess[str] = subprocess.run(
    [
        "bash",
        "-lc",
        "yarn node -list -all 2>/dev/null | awk '$2 == \"RUNNING\" {count++} END {print count + 0}'",
    ],
    capture_output=True,
    text=True,
    check=True,
)
running_nodemanagers: int = int(result.stdout.strip() or 0)
print(f"NodeManagers en estado RUNNING: {running_nodemanagers}")

assert running_nodemanagers >= 2, "Se esperan al menos dos NodeManagers RUNNING"

import os

requested_executors: int = int(os.environ.get("TCDM_SPARK_EXECUTORS", running_nodemanagers))
print(f"Solicitando {requested_executors} ejecutores Spark a YARN.")

spark_submit_cmd: str = f"""/opt/tcdm/venv/bin/spark-submit \
    --master yarn \
    --deploy-mode client \
    --num-executors {requested_executors} \
    --executor-cores 1 \
    --executor-memory 512m \
    --conf spark.executor.memoryOverhead=256m \
    --conf spark.yarn.am.cores=1 \
    --conf spark.yarn.am.memory=256m \
    --conf spark.yarn.am.memoryOverhead=256m \
    --conf spark.scheduler.minRegisteredResourcesRatio=1.0 \
    --conf spark.scheduler.maxRegisteredResourcesWaitingTime=60s \
    --conf spark.pyspark.python=python3 \
    --conf spark.pyspark.driver.python=/opt/tcdm/venv/bin/python \
    /tmp/s4_yarn_job.py"""

print(spark_submit_cmd)

!export HADOOP_CONF_DIR=/opt/bd/hadoop/etc/hadoop && {spark_submit_cmd}

!hdfs dfs -cat /user/luser/s4/yarn-job/business-result/part-*.csv

import csv
import io
from decimal import Decimal

yarn_csv: str = subprocess.run(
    ["hdfs", "dfs", "-cat", "/user/luser/s4/yarn-job/business-result/part-*.csv"],
    capture_output=True,
    text=True,
    check=True,
).stdout

yarn_by_year: dict[int, tuple[int, int, Decimal]] = {
    int(row["d_year"]): (
        int(row["pedidos_distintos"]),
        int(row["lineas_de_venta"]),
        Decimal(row["beneficio_neto_total"]),
    )
    for row in csv.DictReader(io.StringIO(yarn_csv))
}
local_by_year: dict[int, tuple[int, int, Decimal]] = {
    row["d_year"]: (row["pedidos_distintos"], row["lineas_de_venta"], row["beneficio_neto_total"])
    for row in sales_by_year_local_rows
}

assert yarn_by_year == local_by_year, "El resultado sobre YARN no coincide con el local"
print(f"Local y YARN coinciden en los {len(local_by_year)} años.")

!hdfs dfs -cat /user/luser/s4/yarn-job/executor-report/part-*.csv

!yarn application -list -appStates ALL

from subprocess import CompletedProcess

report_result: CompletedProcess[str] = subprocess.run(
    ["bash", "-lc", "hdfs dfs -cat /user/luser/s4/yarn-job/executor-report/part-*.csv"],
    capture_output=True,
    text=True,
    check=True,
)
report_lines: list[str] = [line for line in report_result.stdout.splitlines()[1:] if line.strip()]
executor_hosts: set[str] = {line.split(",")[1] for line in report_lines}

print(f"Hosts que procesaron al menos una partición: {sorted(executor_hosts)}")
print(f"Ejecutores remotos que trabajaron: {len(executor_hosts)}")
assert len(executor_hosts) >= 2, "Se esperan al menos dos hosts remotos distintos"

Concepto	Significado en esta sesión
HDFS o S3	Sistema de almacenamiento que contiene los datos
Parquet	Formato columnar de los ficheros
Spark	Motor que construye y ejecuta el procesamiento
YARN	Gestor que asigna recursos a la aplicación Spark
Partición de Spark	Parte lógica del trabajo que puede ejecutar una tarea
Partición Hive	Organización física de datos por valores de columnas; se verá en S5
Catálogo	Servicio que da nombres y metadatos a las tablas; se verá en S6 y S7

Pestaña	Qué muestra	Para qué sirve en esta sesión
Jobs	Un job por cada acción (`show`, `count`, una escritura…), con su estado, su duración y una línea de tiempo	Comprobar que una transformación no lanza nada y que una acción sí
Stages	Las etapas de cada job y sus tareas: cuántas son, qué ejecutor hizo cada una, cuánto tardó y cuántos datos leyó o intercambió en el shuffle. «DAG Visualization» dibuja el grafo de dependencias entre las operaciones de la etapa	Ver dónde hay un shuffle (dependencia ancha) y cuántas tareas genera cada número de particiones
Storage	Los DataFrames persistidos con `cache()` o `persist()`, qué fracción está materializada y cuánto ocupa en memoria y en disco	Sección de persistencia
Environment	La configuración efectiva: propiedades de Spark, versiones de Java y Scala, rutas de bibliotecas	Confirmar valores como `spark.sql.shuffle.partitions`
Executors	Los ejecutores registrados, con sus núcleos, su memoria, las tareas completadas y el tiempo de recolección de basura. En `local[2]` sólo aparece `driver`	Sobre YARN, ver un ejecutor por contenedor y en qué DataNode está cada uno
SQL / DataFrame	Una entrada por consulta ejecutada, con el plan físico dibujado como grafo y las métricas reales de cada operador (filas producidas, ficheros y bytes leídos, registros de shuffle). «Details» muestra los planes lógico y físico en texto	Es la versión gráfica, y con medidas, de `explain("formatted")`

Sesión 4 — Spark como motor de procesamiento distribuido¶

Diapositivas de la sesión¶

Sesión 4: Spark como motor de procesamiento distribuido

Cómo vamos a recorrer la sesión

Capas que se deben distinguir¶

Objetivos de la sesión¶

Cómo se evalúa esta sesión

Antes de empezar¶

Instalar PySpark en este notebook¶

El motor Spark

El motor Spark¶

API estructurada frente a RDDs¶

Driver, ejecutores y gestor de recursos¶

Términos que reaparecerán en los planes de ejecución¶

Crear una SparkSession local¶

La interfaz web de Spark¶

Primera lectura de Parquet¶

Spark también lee los Parquet de S2

Transformaciones, acciones y planificación¶

Seleccionar pocas columnas frente a select *¶

Filtrado, ordenación y agrupamiento¶

rollup y cube¶

El API de DataFrames en acción

Joins¶

Tipos de join¶

Broadcast join¶

Funciones escalares y agregadas¶

Tipos complejos: arrays y structs¶

Funciones definidas por el usuario (UDFs)¶

Joins, tipos complejos y UDFs

Funciones de ventana¶

Máximo por partición: el pedido mayor de cada cliente¶

lag/lead: días entre pedidos consecutivos de un mismo cliente¶

Media móvil de los últimos 3 pedidos por cliente¶

Suma acumulada del gasto por cliente¶

Las mismas preguntas, ahora en SQL

Spark SQL con vistas temporales¶

Cliente con más pedidos¶

Provincia con más compras¶

Distribución horaria de las compras¶

Las preguntas de S2, ahora con Spark SQL

Beneficio neto por año¶

Escribir y paralelizar con criterio

Guardar resultados¶

El problema de los ficheros pequeños¶

Consolidar antes de escribir con coalesce¶

Particiones de ejecución¶

Reparticionar antes de un join¶

Configuración del shuffle¶

Adaptive Query Execution (AQE)¶

Particiones y shuffle

Persistencia

Persistencia¶

De local[2] a los DataNodes

Driver, ejecutores y YARN¶

Qué Python usan el driver y los ejecutores¶

El script que se enviará a YARN¶

Cuántos ejecutores pedir¶

El envío con spark-submit¶

Comprobar las evidencias¶

El trabajo ocurrió en los DataNodes

Hilo de negocio con TPC-DS¶

Preguntas para interpretar la experiencia¶

Evidencias para la siguiente 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 4

Siguiente paso¶

Seleccionar pocas columnas frente a `select *`¶

`rollup` y `cube`¶

`lag`/`lead`: días entre pedidos consecutivos de un mismo cliente¶

Consolidar antes de escribir con `coalesce`¶

De `local[2]` a los DataNodes

El envío con `spark-submit`¶