Módulo 2: Conectividad y Transformación¶

La capacidad de analizar datos a gran escala depende críticamente de cómo movemos la información desde su origen hacia nuestro motor de cómputo. En este módulo, exploramos el patrón de Ingestion Pipeline y el impacto de la latencia de E/S en el rendimiento.

El cuello de botella: IO Latency¶

En la arquitectura de datos moderna, uno de los mayores desafíos es la IO Latency (latencia de Entrada/Salida). Cuando extraemos datos de un RDBMS tradicional (como MariaDB o MySQL), nos enfrentamos a una discrepancia fundamental de formatos:

RDBMS (Row-based): Las bases de datos relacionales almacenan los datos en filas. Esto es ideal para transacciones (OLTP), donde necesitamos leer o escribir un registro completo.
Análisis (Columnar): Para el análisis de datos (OLAP), necesitamos leer columnas enteras para calcular agregaciones (sumas, promedios). Leer filas completas cuando solo necesitamos una columna genera una lectura innecesaria de datos, saturando el ancho de banda de E/S.

El reto consiste en mover datos desde un formato basado en filas hacia un formato columnar de manera eficiente, minimizando el tiempo de transferencia y la CPU dedicada a la serialización.

Enfoque en Apache Arrow¶

Para resolver este problema, Polars utiliza Apache Arrow como su núcleo de memoria. Arrow no es solo un formato de almacenamiento, sino un estándar de memoria columnar.

¿Cómo gestiona Polars la ingesta?¶

Cuando Polars lee datos de una base de datos a través de conectores optimizados (como connectorx), el proceso ocurre de la siguiente manera:

Stream de Filas \(\rightarrow\) Arrays Columnares: El conector extrae el flujo de filas del RDBMS y, en lugar de convertirlas en objetos Python lentos, las organiza inmediatamente en arrays contiguos de memoria siguiendo la especificación de Arrow.
Zero-copy: Al utilizar Arrow, Polars puede pasar estos datos entre diferentes herramientas o etapas del pipeline sin necesidad de copiar los datos en memoria, eliminando el coste de serialización/deserialización.
Eficiencia de CPU: Al estar los datos alineados en columnas, el procesador puede aplicar instrucciones SIMD (Single Instruction, Multiple Data) para procesar miles de valores en un solo ciclo de reloj.

Implementación del Ingestion Pipeline (RDBMS 👉 Polars)¶

A continuación, implementamos el patrón básico de ingesta: extraer datos de MariaDB y cargarlos en un DataFrame de Polars para su análisis.

1. Preparación del entorno¶

Necesitamos instalar polars y connectorx (un conector escrito en Rust extremadamente rápido para cargar datos de SQL a Arrow).

pip install polars connectorx

2. Lectura optimizada vía ConnectorX¶

connectorx es capaz de leer la base de datos y producir directamente una tabla Arrow, evitando el cuello de botella de Python.

import polars as pl
import connectorx as cx
import pyarrow.compute as pc

# Configuración de conexión
conn = 'mysql://usuario:contraseña@mariadb:3306/kafka'
query = 'SELECT * FROM songs'

# Leer produciendo una tabla Arrow directamente en memoria
ta = cx.read_sql(conn, query, return_type='arrow')
print("Esquema de la Tabla Arrow:", ta.schema)

# Ejemplo de filtrado rápido sobre la tabla Arrow antes de pasar a Polars
pjta = ta.filter(pc.equal(ta['artist'], 'Pearl Jam'))
print(pjta.select(["album","title"]))

3. Ingesta directa a Polars DataFrame¶

Para la mayoría de los casos, la función read_database_uri es la forma más sencilla y eficiente de implementar este pipeline, ya que abstrae la gestión de Arrow internamente.

import polars as pl

uri = "mysql://usuario:contraseña@mariadb:3306/kafka"
query = "SELECT * FROM songs"

# Ingesta directa: RDBMS -> Polars DataFrame
df = pl.read_database_uri(query=query, uri=uri)
print(df.head())

4. Conectividad mediante SQLAlchemy¶

En entornos donde se requiere una gestión más robusta de las conexiones (pooling, transacciones), podemos utilizar SQLAlchemy.

from sqlalchemy import create_engine
import polars as pl

# Configuración de conexión
username = 'usuario'
password = 'contraseña'
host = 'mariadb:3306'
database = 'kafka'

# Creación del motor (Engine) con pool de conexiones
dbcon = create_engine(
    f'mariadb+mariadbconnector://{username}:{password}@{host}/{database}',
    pool_size=10, 
    max_overflow=20
)

# Consultar la base de datos usando la conexión de SQLAlchemy
df = pl.read_database(query="SELECT * FROM songs", connection=dbcon)

Escritura de datos en la Base de Datos¶

El flujo inverso también es posible. Podemos tomar un dataset procesado en Polars y volcarlo en un RDBMS para persistir los resultados.

# 1. Leer datos de una fuente local (ej. CSV)
dfpl = pl.read_csv("Eurostar/stations.csv")

# 2. Volcar el DataFrame columnar en la tabla de la base de datos (Row-based)
dfpl.write_database(
    table_name='stations', 
    connection=dbcon, 
    engine='sqlalchemy', 
    if_table_exists="replace"
)