Ecosistema Apache

Hadoop y HDFS

Apache Hadoop fue el sistema que democratizó el Big Data entre 2006 y 2018. Aunque hoy está siendo reemplazado por soluciones cloud, entender su arquitectura es fundamental para comprender cómo evolucionó la industria.

La Arquitectura de Hadoop

HDFS: Conceptos Clave

• Bloques de 128MB (default): HDFS divide los archivos en bloques grandes. Un archivo de 1GB se divide en ~8 bloques distribuidos por el cluster.
• Replicación (default 3x): Cada bloque se replica en 3 DataNodes diferentes para tolerancia a fallos. Si un nodo falla, los datos siguen disponibles.
• Write-once, read-many: HDFS está optimizado para lecturas grandes. No soporta modificaciones parciales de archivos.
• NameNode - Single Point of Failure: El NameNode guarda todos los metadatos del filesystem. Su falla detiene el cluster. Hadoop 2.x introdujo NameNode HA para resolver esto.
• Data Locality: YARN intenta ejecutar el procesamiento en el mismo nodo donde están los datos (o en un nodo cercano), minimizando transferencia de red.

# ── COMANDOS HDFS ESENCIALES ───────────────────────────────────────────────
# HDFS usa una API similar a Linux filesystem

# Listar archivos
hdfs dfs -ls /user/de/data/
hdfs dfs -ls -R /warehouse/  # Recursivo

# Subir/bajar archivos
hdfs dfs -put local_file.csv /user/de/data/
hdfs dfs -get /user/de/data/output.parquet ./local_output.parquet

# Ver contenido
hdfs dfs -cat /user/de/data/file.csv | head -20
hdfs dfs -tail /user/de/logs/app.log

# Crear directorios y gestión
hdfs dfs -mkdir -p /warehouse/database/table/
hdfs dfs -cp /source/path /destination/path
hdfs dfs -mv /old/path /new/path
hdfs dfs -rm -r /path/to/delete/

# Información sobre archivos y bloques
hdfs dfs -du -h /warehouse/  # Disk usage human-readable
hdfs fsck /warehouse/orders/ -files -blocks  # Verificar integridad
hdfs dfsadmin -report  # Estado del cluster

# Configuración de replicación
hdfs dfs -setrep -w 2 /user/de/large-table/  # Reducir a 2 réplicas

Apache Hive

Hive convirtió HDFS en un Data Warehouse al agregar una capa SQL sobre MapReduce/Spark. Fue la primera solución que permitió a analistas consultar petabytes de datos con SQL estándar.

-- ── HIVE SQL: Crear tablas sobre archivos HDFS ─────────────────────────────
-- Tabla externa: Hive solo define el esquema, los datos quedan en HDFS
CREATE EXTERNAL TABLE IF NOT EXISTS sales (
    sale_id     BIGINT,
    customer_id INT,
    amount      DOUBLE,
    sale_date   STRING
)
ROW FORMAT DELIMITED
    FIELDS TERMINATED BY ','
    LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '/warehouse/raw/sales/'
TBLPROPERTIES ("skip.header.line.count"="1");

-- Tabla particionada y en formato columnar (mucho más eficiente)
CREATE EXTERNAL TABLE IF NOT EXISTS sales_parquet (
    sale_id     BIGINT,
    customer_id INT,
    amount      DOUBLE
)
PARTITIONED BY (year INT, month INT)  -- Partition columns
STORED AS PARQUET
LOCATION '/warehouse/sales_parquet/'
TBLPROPERTIES ("parquet.compression"="SNAPPY");

-- Agregar partición manualmente o usar MSCK REPAIR
ALTER TABLE sales_parquet ADD PARTITION (year=2026, month=1)
    LOCATION '/warehouse/sales_parquet/year=2026/month=1/';

-- MSCK REPAIR: descubrir particiones automáticamente
MSCK REPAIR TABLE sales_parquet;

-- ── QUERIES HIVE OPTIMIZADAS ──────────────────────────────────────────────
-- Siempre filtrar por partición para aprovechar partition pruning
SELECT customer_id, SUM(amount) AS total
FROM sales_parquet
WHERE year = 2026 AND month BETWEEN 1 AND 3  -- ✅ Usa partición
GROUP BY customer_id;

-- ORC vs Parquet en Hive
-- ORC (Optimized Row Columnar): mejor para Hive, soporta ACID, mejor compresión
-- Parquet: mejor interoperabilidad con Spark, Python, Arrow

Formatos de Archivo en Big Data

La elección del formato de archivo impacta directamente el costo, la velocidad y la flexibilidad de tu plataforma de datos. Este es un conocimiento fundamental que diferencia a los seniors.

# Escribir Parquet con Python
import pandas as pd
df.to_parquet('output.parquet', engine='pyarrow', compression='snappy')

# Leer solo columnas necesarias (columnar advantage)
df = pd.read_parquet('output.parquet', columns=['customer_id', 'amount'])

// Schema Avro (JSON)
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "long"},
    {"name": "amount",   "type": "double"},
    {"name": "status",   "type": "string", "default": "pending"},
    // Nuevo campo con default: schema-compatible!
    {"name": "discount", "type": ["null", "double"], "default": null}
  ]
}

Comparativa de Formatos

Delta Lake

Delta Lake es una capa de storage open source que agrega confiabilidad ACID sobre Parquet en object storage (S3, GCS, ADLS). Es el componente central del patrón Lakehouse.

Funcionalidades Clave de Delta Lake

"""
Delta Lake con PySpark
"""
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *

spark = (SparkSession.builder
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)

# ── CREAR DELTA TABLE ──────────────────────────────────────────────────────
df = spark.read.parquet("s3://bucket/raw/orders/")
(df.write.format("delta")
    .mode("overwrite")
    .partitionBy("year", "month")
    .option("overwriteSchema", "true")
    .save("s3://bucket/silver/orders/")
)

# ── TIME TRAVEL ────────────────────────────────────────────────────────────
# Ver historial de la tabla
spark.sql("DESCRIBE HISTORY delta.`s3://bucket/silver/orders/`").show()

# Consultar versión específica
df_v3 = spark.read.format("delta").option("versionAsOf", 3).load("s3://bucket/silver/orders/")

# Consultar por timestamp
df_yesterday = spark.read.format("delta") \
    .option("timestampAsOf", "2026-01-14") \
    .load("s3://bucket/silver/orders/")

# ── MERGE (CDC Pattern) ────────────────────────────────────────────────────
delta_table = DeltaTable.forPath(spark, "s3://bucket/silver/orders/")

# CDC: aplicar cambios incrementales
cdc_df = spark.read.parquet("s3://bucket/cdc/orders/latest/")

(delta_table.alias("target")
    .merge(cdc_df.alias("source"), "target.order_id = source.order_id")
    .whenMatchedUpdate(
        condition="source.op = 'UPDATE'",
        set={"status": "source.status", "updated_at": "source.updated_at"}
    )
    .whenMatchedDelete(condition="source.op = 'DELETE'")
    .whenNotMatchedInsert(
        condition="source.op = 'INSERT'",
        values={"order_id": "source.order_id", "amount": "source.amount", "status": "source.status"}
    )
    .execute()
)

# ── OPTIMIZE & VACUUM ──────────────────────────────────────────────────────
# Compactar archivos pequeños (el "small files problem")
spark.sql("OPTIMIZE delta.`s3://bucket/silver/orders/` ZORDER BY (customer_id)")
# ZORDER: co-localiza datos relacionados → data skipping más efectivo

# Limpiar versiones históricas antiguas (>7 días default)
spark.sql("VACUUM delta.`s3://bucket/silver/orders/` RETAIN 168 HOURS")

Apache Iceberg

Apache Iceberg es el principal competidor de Delta Lake. Iniciado por Netflix, ofrece ACID sobre object storage con un enfoque más abierto y con mejor soporte multi-engine.

Delta Lake vs Apache Iceberg vs Apache Hudi