Pipeline de Datos con Apache Arrow

import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq
import pyarrow.dataset as ds
import pyarrow.ipc as ipc

# ── 1. Construir tabla Arrow desde datos TechGadgetsPro ─────────────────────
events = pa.table({
    "event_id":         pa.array([10000001, 10000002, 10000007, 10000008, 10000023], type=pa.int64()),
    "user_id":          pa.array([7823491,  7823491,  7823491,  7823491,  9041122],  type=pa.int64()),
    "event_type":       pa.array(["page_view", "add_to_cart", "checkout", "purchase", "purchase"]),
    "product_id":       pa.array(["SKU-0041", "SKU-0041", "SKU-0041", "SKU-0041", "SKU-0093"]),
    "product_category": pa.array(["electronics", "electronics", "electronics", "electronics", "electronics"]),
    "revenue":          pa.array([0.00, 0.00, 0.00, 129.99, 59.90],           type=pa.float64()),
    "year":             pa.array([2026] * 5, type=pa.int32()),
    "month":            pa.array([6]    * 5, type=pa.int32()),
})
# En producción se leen 2.14M filas desde Parquet/S3 — misma API

# ── 2. KPIs con compute (vectorizado, sin bucles Python) ────────────────────
purchases = pc.filter(events, pc.equal(events["event_type"], "purchase"))

total_revenue    = pc.sum(purchases["revenue"]).as_py()          # 189.89
avg_ticket       = pc.mean(purchases["revenue"]).as_py()         # 94.95
conversion_rate  = purchases.num_rows / events.num_rows          # 0.40 (40%)

# Top productos por revenue — DuckDB sobre Arrow (zero-copy)
import duckdb
top_products = duckdb.arrow(purchases).query("""
    SELECT product_id, product_category,
           COUNT(*) AS purchases,
           SUM(revenue) AS total_revenue
    FROM purchases
    GROUP BY product_id, product_category
    ORDER BY total_revenue DESC
    LIMIT 5
""").arrow()   # resultado también es Arrow table

print(f"Revenue: €{total_revenue:,.2f} | Ticket: €{avg_ticket:.2f} | Conv.: {conversion_rate:.1%}")

# ── 3. Escribir dataset particionado (Hive-style) en S3 ────────────────────
ds.write_dataset(
    events,
    "s3://dataflow-techgadgetspro/events/",
    format="parquet",
    partitioning=ds.partitioning(
        pa.schema([("year", pa.int32()), ("month", pa.int32())]),
        flavor="hive",   # → year=2026/month=6/part-0.parquet
    ),
    existing_data_behavior="overwrite_or_ignore",
    file_options=pq.ParquetFileFormat().make_write_options(compression="zstd"),
)
# Resultado: 3 GB CSV  ──►  214 MB Parquet/ZSTD  (14× menor en disco)

# ── 4. Lectura con partition pruning + predicate pushdown ──────────────────
dataset = ds.dataset(
    "s3://dataflow-techgadgetspro/events/",
    format="parquet",
    partitioning="hive",
)

june_purchases = dataset.scanner(
    columns=["user_id", "product_id", "revenue", "timestamp"],
    filter=(
        (ds.field("year") == 2026) &
        (ds.field("month") == 6) &
        (ds.field("event_type") == "purchase") &
        (ds.field("revenue") > 0)
    ),
).to_table()
# Sólo lee los archivos de year=2026/month=6 — los demás se saltan completamente
# Lectura selectiva de columnas: I/O reducido 4× respecto a leer todas

# Streaming para lotes de 100K filas (útil con 2M eventos/día)
pf = pq.ParquetFile("local_events.parquet")
for batch in pf.iter_batches(batch_size=100_000):
    filtered = pc.filter(batch, pc.greater(batch["revenue"], 0))
    process_batch(filtered)   # RAM constante ~200MB en lugar de 3 GB