7 Python ETL eszköz adatmérnököknek kódpéldákkal

Az adatfolyamatok (ETL: Extract, Transform, Load) építése az adatmérnökök egyik legfontosabb feladata. Bár tisztán Python és Pandas használatával is lehetséges, a specializált eszközök jobban kezelik az ütemezést, hibakezelést és skálázhatóságot. A választás azonban nehéz lehet, mert egyes eszközök túl bonyolultak, mások pedig korlátozottak. Ez a cikk hét olyan Python alapú ETL eszközt mutat be, amelyek aktív közösséggel rendelkeznek és éles környezetben is megállják a helyüket. A cikkben a bemutatás mellett, minta kódokat és a mélyebb tanuláshoz szükséges forrásokat is elérhetővé teszünk.

1. Apache Airflow, a munkafolyamatok karmestere

Amikor a szkriptek már túl bonyolultak, szükség van egy karmesterre. Az Apache Airflow az iparági szabvány a munkafolyamatok programozott létrehozására és felügyeletére.

Főbb jellemzői:

A folyamatokat irányított aciklikus gráfként (DAG) definiálhatod Python kódban.
Erős webes felületet biztosít a folyamatok figyelésére és a hibák keresésére.
Rengeteg beépített operátorral rendelkezik adatbázisokhoz és API-khoz.

Így néz ki egy egyszerű irányított gráf vagyis DAG definíciója:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def feladat():
    print("Adatok feldolgozása...")

with DAG('elso_dag', start_date=datetime(2025, 1, 1)) as dag:
    task1 = PythonOperator(
        task_id='feldolgozas',
        python_callable=feladat
    )

Ha bonyolult függőségeket kell kezelned akkor ez a legjobb választás.

További információ: Airflow hivatalos dokumentáció

2. Luigi, az egyszerűség ereje

Ha az Airflow túl soknak tűnik, a Luigi jó alternatíva lehet. Ezt a Spotify fejlesztette ki kifejezetten hosszú futású kötegelt (batch) folyamatokhoz.

Miért érdemes használni:

Osztályalapú megközelítést alkalmaz, ahol minden feladat egy Python osztály.
Automatikusan kezeli a függőségeket.
Kisebb csapatok számára könnyebben fenntartható.

A kód felépítése objektumorientált szemléletet követ:

import luigi

class AdatLetoltes(luigi.Task):
    def output(self):
        return luigi.LocalTarget('adat.csv')

    def run(self):
        with self.output().open('w') as f:
            f.write('id,nev\n1,Teszt')

if __name__ == '__main__':
    luigi.run()

Kisebb csapatoknak ideális a könnyű karbantarthatóság miatt.

További információ: Luigi GitHub oldal

3. Prefect, a modern megközelítés

A Prefect a modern adatmérnökök választása. Könnyebb megtanulni, mint az Airflow-t, de képes kezelni a nagy méretű rendszereket is.

Előnyei:

Szabványos Python függvényeket használ dekorátorokkal.
Kiváló hibakezelést és automatikus újrapróbálkozást kínál.
Felhő alapú és saját szerveres verziója is van.

Nézd meg mennyire letisztult a szintaxis:

from prefect import flow, task

@task
def adat_kinyeres():
    return [1, 2, 3]

@task
def transzformacio(adat):
    return [x * 2 for x in adat]

@flow
def etl_folyamat():
    nyers = adat_kinyeres()
    transzformacio(nyers)

if __name__ == "__main__":
    etl_folyamat()

Kiváló hibakezelést és automatikus újrapróbálkozást kínál külön konfiguráció nélkül.

További információ: Prefect dokumentáció

4. Dagster, adatközpontú tervezés

A Dagster nem feladatokra, hanem adatvagyonra (data assets) fókuszál. A deklaratív megközelítés miatt könnyebb átlátni az adat útját a rendszerben.

Kiemelt funkciók:

Szoftverdefiniált eszközök (Software-defined Assets) használata.
Beépített tesztelési eszközök és lokális fejlesztői környezet.
Világos adatkapcsolati térkép (data lineage).

A kód így garantálja a típusbiztonságot:

from dagster import asset

@asset
def nyers_adatok():
    return [1, 2, 3]

@asset
def feldolgozott_adatok(nyers_adatok):
    return [x + 1 for x in nyers_adatok]

A beépített tesztelési eszközök miatt nagyon stabil rendszereket építhetsz vele.

További információ: Dagster.io

5. PySpark, ha nagyban kell játszani

A hatalmas adathalmazok feldolgozásához elosztott rendszer kell. A PySpark az Apache Spark Python felülete, amely lehetővé teszi a klasztereken történő számítást.

Miért elengedhetetlen:

Automatikusan elosztja a feladatokat több gép között.
Magas szintű API-t nyújt az adattranszformációkhoz.
Támogatja a kötegelt és a valós idejű (streaming) feldolgozást is.

Így olvass be és szűrj gigabájtnyi adatot:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ETL").getOrCreate()

df = spark.read.csv("nagy_adat.csv", header=True)
szurt_df = df.filter(df['ar'] > 1000)

szurt_df.show()

Ez az eszköz elengedhetetlen a Big Data világában.

További információ: PySpark útmutató

6. Mage AI, a notebookok utódja

A Mage AI ötvözi a notebookok egyszerűségét a gyártási rendszerek stabilitásával. Ideális azoknak, akik gyorsan akarnak prototípusból éles rendszert építeni.

Erősségei:

Interaktív fejlesztői felület.
Beépített blokkok az adatforrásokhoz és célállomásokhoz.
Átlátható felület a hibakereséshez.

Egy tipikus adatbetöltő blokk így néz ki:

import pandas as pd

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader

@data_loader
def load_data(*args, **kwargs):
    url = 'https://pelda.hu/adat.csv'
    return pd.read_csv(url)

Gyors prototípusok építésére ez a leggyorsabb eszköz.

További információ: Mage AI dokumentáció

7. Kedro, szoftvermérnöki alapok

A Kedro segít rendszert vinni a káoszba. Ez a keretrendszer szoftvermérnöki elveket alkalmaz az adatfolyamatokra, így a kód fenntartható és tesztelhető marad.

Jellemzők:

Szabványosított projektstruktúrát kényszerít ki.
Beépített adatkatalógussal kezeli a fájlútvonalakat.
Jól integrálható az Airflow-val vagy Prefect-tel.

A node és a pipeline kapcsolata így épül fel:

from kedro.pipeline import node, pipeline

def tisztitas(adat):
    return adat.dropna()

def create_pipeline(**kwargs):
    return pipeline([
        node(
            func=tisztitas,
            inputs="nyers_adat",
            outputs="tiszta_adat",
            name="tisztito_node"
        )
    ])

Nagyvállalati környezetben ez a struktúra aranyat ér.

További információ: Kedro dokumentáció

Összehasonlító táblázat

A táblázat segít a gyors döntésben a feladat típusa alapján.

Eszköz	Fő erősség	Kinek ajánlott
Airflow	Ütemezés	Nagy csapatoknak
Luigi	Függőségek	Egyszerűbb projektekhez
Prefect	Modern szintaxis	Python fejlesztőknek
Dagster	Adatbiztonság	Akiknek a minőség a fő
PySpark	Skálázhatóság	Big Data mérnököknek
Mage AI	Gyorsaság	Kezdőknek és startupoknak
Kedro	Struktúra	Szoftvermérnököknek