Módulo 3: Arrow + DuckDB

El cuello de botella: Copy Overhead¶

En el ecosistema de datos tradicional, cada herramienta (Pandas, Spark, SQL engines) tiene su propia representación interna de los datos en memoria. Cuando necesitamos mover datos de una herramienta a otra (por ejemplo, de un extractor SQL a un procesador de DataFrames), ocurre un proceso costoso llamado Copy Overhead.

El Copy Overhead implica:
1. Serialización: Convertir el formato interno de la herramienta A a un formato intercambiable.
2. Transporte: Mover los bytes a través de la memoria.
3. Deserialización: Reconstruir los datos en el formato interno de la herramienta B.

En conjuntos de datos masivos, este proceso de "copia y conversión" se convierte en el principal cuello de botella, consumiendo CPU y duplicando el uso de memoria RAM, lo que a menudo provoca errores de Out of Memory (OOM).

Enfoque en Apache Arrow: El estándar de Zero-copy¶

Para eliminar este problema, nace Apache Arrow. Arrow no es una base de datos ni un lenguaje, sino una especificación de formato de memoria columnar.

¿Qué es el Zero-copy?¶

El concepto de Zero-copy permite que diferentes herramientas accedan a la misma región de memoria sin necesidad de duplicar o transformar los datos. Si Polars y DuckDB utilizan el estándar Arrow, pueden "compartir" la memoria:

• Memoria Compartida: Los datos se almacenan en un layout columnar estandarizado.
• Acceso Directo: Polars puede crear un DataFrame que apunta directamente a los buffers de memoria que DuckDB acaba de procesar, y viceversa.
• Eficiencia: El tiempo de transferencia es prácticamente cero, independientemente del tamaño del dataset.

Esto transforma la arquitectura de datos de un modelo de "pasamanos" (copiando datos) a un modelo de "ventana" (donde cada herramienta mira los mismos datos desde su propia perspectiva).

Orquestación Cross-tool: Polars ↔️ DuckDB¶

La verdadera potencia surge al combinar la capacidad de manipulación de Polars con la potencia de consulta SQL de DuckDB, utilizando Arrow como el tejido conectivo.

Ejemplo: Integración de Alto Rendimiento¶

En este patrón, utilizamos Polars para la ingesta y limpieza (aprovechando su Lazy Evaluation) y DuckDB para análisis complejos mediante SQL, todo sin salir del ecosistema Arrow.

import polars as pl
import duckdb

# 1. Ingesta perezosa con Polars (Apache Arrow bajo el capó)
schema = {
    "StopCode": pl.String,
    "StopName": pl.String,
    "Country": pl.String,
    "StopLat": pl.Float64,
    "StopLng": pl.Float64,
}

stations_csv = "https://drive.upm.es/public.php/dav/files/X99My74a3LKyAK4"
lazy_stations = pl.scan_csv(
    stations_csv,
    schema=schema,
    null_values=["", "NA", "null"]
)

# Ejecutamos la computación y obtenemos un DataFrame de Polars (formato Arrow)
df_stations = lazy_stations.collect()

# 2. Interoperabilidad Zero-copy con DuckDB
# DuckDB puede consultar directamente el objeto DataFrame de Polars sin copiarlo
res = duckdb.sql("""
    SELECT StopName, Country 
    FROM df_stations 
    WHERE Country = 'Germany' 
    LIMIT 5
""").pl() # Convertimos el resultado de vuelta a Polars (también Zero-copy)

print(res)

Casos de Uso: Joins entre datos externos y bases de datos¶

A continuación, vemos cómo aplicar este flujo de interoperabilidad para resolver consultas complejas combinando fuentes heterogéneas.

Join entre Parquet y CSV alojados en la Web¶

Q1: Horarios de salida en orden temporal creciente de los servicios Eurostar en cada estación la semana 13.

# WebDAV UPMdrive
stations_csv = "https://drive.upm.es/public.php/dav/files/X99My74a3LKyAK4"
eurotrain_parquet = "https://drive.upm.es/public.php/dav/files/XTzftsPGJMARfDj"

# Definición de esquema para el CSV
schema = {
    "StopCode": pl.String, "StopName": pl.String, "Country": pl.String,
    "StopLat": pl.Float64, "StopLng": pl.Float64,
}

# Construcción de la consulta perezosa (LazyFrame)
# Polars optimiza el plan de ejecución antes de tocar los datos
join_lazy = (
    pl.scan_parquet(eurotrain_parquet)
    .filter((pl.col("ServiceType") == 'Eurostar') & (pl.col("ns") == 13) & (pl.col("StopDepartureTime").is_not_null()))
    .join(pl.scan_csv(stations_csv, schema=schema), on="StopCode")
    .filter(pl.col("StopName") == 'Rotterdam Centraal')
    .select(["ServiceDate", "ServiceID", "StopArrivalTime", "StopDepartureTime"])
    .sort("ServiceID", "StopArrivalTime", "StopDepartureTime")
)

# Ejecución final: aquí es donde ocurre la lectura y el join optimizado
res = join_lazy.collect()
print(res)

Join entre tabla MariaDB y CSV alojado en la Web¶

Q2: Itinerario (secuencia de estaciones en orden temporal creciente) de un servicio concreto (13224525).

# Conexión con la base de datos
uri = "mysql://eurostar:eurostar@mariadb:3306/eurostar"
sql_query = "SELECT tts, StopCode, ServiceDate, ServiceType, ServiceID, StopArrivalTime, StopDepartureTime FROM eurostar.EuroTrain"

# Orquestación:
# 1. Leemos de MariaDB y pasamos a modo Lazy
# 2. Hacemos Join con el CSV remoto usando Arrow
# 3. Ordenamos y filtramos
lazy_join = (
    pl.read_database_uri(query=sql_query, uri=uri).lazy()
    .filter(pl.col("ServiceID") == '13224525')
    .join(pl.scan_csv(stations_csv, schema=schema), on="StopCode")
    .sort("tts")
    .select(["StopName", "ServiceDate", "ServiceType", "StopArrivalTime", "StopDepartureTime"])
)

res = lazy_join.collect()
print(res)