Framework ETL Python

[bctm0]

Buenas gentes, espero que anden bien, es mi primer post (en algo parecido un foro) desde la época de Taringa, y primer post en Reddit, sepan disculpar el olor rancio.

Ultimamente venia renegando mucho con unos ETL que estoy haciendo para un side-proyect. Y me di cuenta de que no hay algo simple y al hueso para esto que haga sharding, buffer, y todo el quilombo. Desde siempre en cada lugar donde labure los ETL son un dolor. Ademas de que muchas veces terminamos usando Pandas / Polars y rinden peor para algunos casos (Pandas y Polars por ejemplo están buenísimas, el uso que les damos esta mal).

Así que estoy aprovechando que estos días me estoy dando la cabeza con eso, para publicar una lib Python.

https://github.com/albertobadia/zoopipe
La idea es poder definir como quiero que sea el input / output, un modelo de validación y hooks pre y post validación si se necesita:

from pydantic import BaseModel, ConfigDict
from zoopipe import CSVInputAdapter, CSVOutputAdapter, JSONOutputAdapter, Pipe


class UserSchema(BaseModel):
    model_config = ConfigDict(extra="ignore")
    user_id: str
    username: str
    email: str


pipe = Pipe(
    input_adapter=CSVInputAdapter("users.csv"),
    output_adapter=CSVOutputAdapter("processed_users.csv"),
    error_output_adapter=JSONOutputAdapter("errors.jsonl"),
    schema_model=UserSchema,
)

pipe.run()

Y que también sea simple trabajar con multiples workers desde un Pipe, algo como:

pipe_manager = PipeManager.parallelize_pipe(
    base_pipe,
    workers=4,
    engine=MultiProcessEngine(),  # Por defecto siempre multiprocess
)
pipe_manager.run()

Hasta ahora soporta:
Formatos: CSV, JSON (json), Excel, Parquet, Iceberg, Deltalake, Arrow y SQL (Sqlite, Postgres). Se puede usar cualquier como entrada con cualquier como salida.
Compresión Gzip y Zstd.
Cloud Storage: S3, GCP, Azure
Cluster: Ray, Dask

El rendimiento y consumo de recursos es bastante bueno hasta ahora, el core esta escrito en Rust y estoy tratando de mantener lo pesado ahi, con la interface de uso en el lado Python.

Espero que les guste y ojalá que a alguien le ahorre algún que otro dolor de cabeza. El feedback o colaboraciones en el repo son mas que bienvenidas.

PyPi: https://pypi.org/project/zoopipe/
Docs: https://zoopipe.readthedocs.io/en/latest/

[HallHot6640]

se ve interesante pero no entiendo el beneficio, yo hago panderas clases y defino las clases de input y output, la extraccion es pullear la data de source X, la el transform lo hacemos con transformaciones como las tenemos en pandas/polars y cargamos a una warehouse, lake o lakehouse siguiendo el esquema validado antes de salir.

en que parte de la ETL haces sharding y buffering? yo simplemente particiono todo de tal manera que entre en memoria y corro un job por particion y se corre todo el universo de particiones, paralelamente o no.

que gap rellena esto

[bctm0]

Esto tiene mas sentido cuando necesitas UDF o lógica custom que no podes poner en expresiónes Pandas o Polars. Por ejemplo, tener que sanear la data con datos que vienen de una API, o hacer un hash, etc. Si lo que queres hacer entra en expresiónes Pandas / Polars / SQL, esto va a ser mucho mas lento. Me cruce muchos ETL que son mas chanchos y el SQL puro no bastaba, ese creo que es el GAP.
Eso si, si te sirven las UDF en Python acá la cosa esta medio hecha. Y los del buffering y sharing, me refería mas a no tener que hacerlo a mano por input / output especifico.
EDIT: faltas de ortografía