Top 5 Airflow-alternatieven voor data-orchestratie (met codevoorbeelden)

Lees de blog Airflow vs Prefect: kiezen wat past bij jouw dataworkflow voor een gedetailleerde vergelijking tussen deze twee tools voor data-orchestratie. 

Aan de slag met Prefect

We beginnen ons Prefect-project met het installeren van het Python-pakket. Voer de volgende opdracht uit in een terminal.

$ pip install -U prefect

Daarna maken we een Python-script met de naam prefect_etl.py en schrijven we de volgende code.

from prefect import task, flow
import pandas as pd

# Extract data
@task
def extract_data():
    # Simulating data extraction
    data = {
        "name": ["Alice", "Bob", "Charlie"],
        "age": [25, 30, 35],
        "city": ["New York", "Los Angeles", "Chicago"]
    }
    df = pd.DataFrame(data)
    return df

# Transform data
@task
def transform_data(df: pd.DataFrame):
    # Example transformation: adding a new column
    df["age_plus_ten"] = df["age"] + 10
    return df

# Load data
@task
def load_data(df: pd.DataFrame):
    # Simulating data load
    print("Loading data to target destination:")
    print(df)

# Defining the flow
@flow(log_prints=True)
def etl():
    raw_data = extract_data()
    transformed_data = transform_data(raw_data)
    load_data(transformed_data)

# Running the flow
if __name__ == "__main__":
    etl()

De bovenstaande code definieert de taakfuncties extract_data(), transform_data(), en load_data() en voert ze achter elkaar uit in een flowfunctie genaamd etl(). Deze functies zijn gemaakt met Prefect Python-decorators. 

Kort gezegd maken we een pandas DataFrame, transformeren dat en tonen het eindresultaat met print. Dit is een eenvoudige manier om een ETL-pijplijn te simuleren.

Om de workflow uit te voeren, draai je het Python-script met het volgende commando.

$ python prefect_etl.py 

Zoals je ziet, is onze workflowrun succesvol afgerond.

Prefect flow-run-logs.

De flow deployen

We gaan onze workflow nu deployen, zodat we deze op een schema kunnen draaien of op een event kunnen triggeren. Deployen van de flow maakt het ook mogelijk om meerdere workflows centraal te monitoren en beheren.

Voor het deployen gebruiken we de Prefect CLI. De functie deploy vereist de naam van het Python-bestand, de flowfunctienaam in dat bestand en de deploymentnaam. In dit geval noemen we deze deployment “simple_etl”.

$ prefect deploy prefect_etl.py:etl -n 'simple_etl'

Na het draaien van bovenstaand script in de terminal kun je de melding krijgen dat je geen worker pool hebt om de deployment uit te voeren. Maak de worker pool met het volgende commando.

$ prefect worker start --pool 'datacamp'

Nu we een worker pool hebben, openen we een nieuw terminalvenster en draaien we de deployment. Het commando prefect deployment run vereist “<flow-function-name>/<deployment-name>” als argument, zoals hieronder.

$ prefect deployment run 'etl/simple_etl

Als resultaat van het draaien van de deployment krijg je een melding dat de workflow draait. Meestal krijgt de flowrun die wordt aangemaakt een willekeurige naam, in mijn geval witty-lorikeet.

Creating flow run for deployment 'etl/simple_etl'...
Created flow run 'witty-lorikeet'.
└── UUID: 4e0495b0-9c7e-4ed8-b9ab-5160994dc7f0
└── Parameters: {}
└── Job Variables: {}
└── Scheduled start time: 2024-06-22 14:05:01 PKT (now)
└── URL: <no dashboard available>

Voor de volledige log schakel je terug naar het terminalvenster waar je de worker pool startte.

Samenvatting van Prefect flow-run.

Je moet de Prefect webserver starten om de flowrun overzichtelijker te visualiseren en andere workflows te beheren.

$ prefect server start 

Na uitvoering van het bovenstaande commando word je doorgestuurd naar het Prefect-dashboard. Je kunt ook rechtstreeks naar het adres http://127.0.0.1:4200 in je browser gaan.

Prefect webserver-UI

Met het dashboard kun je de workflow opnieuw draaien, logs bekijken, work pools controleren, notificaties instellen en andere geavanceerde opties selecteren. Het is een complete oplossing voor moderne data-orchestratiebehoeften.

Wil je leren hoe je machinelearning-pijplijnen bouwt en uitvoert met Prefect? Volg dan de tutorial Using Prefect for Machine Learning Workflows.

2. Dagster

Dasgter is een open-source framework voor data-engineers om datapijplijnen te definiëren, plannen en monitoren. Het is zeer schaalbaar en faciliteert samenwerking tussen verschillende datateams. 

Dagster laat gebruikers hun data-assets definiëren als Python-functies met decorators. Zodra deze assets zijn gedefinieerd, kun je ze naadloos uitvoeren via scheduling of event-gebaseerde triggers.

Vergeleken met Airflow kunnen we met Dagster de pijplijn lokaal ontwikkelen, testen en reviewen, biedt het een asset-gebaseerde benadering van orkestratie en is het cloud- en container-native.

Aan de slag met Dagster

We maken een eenvoudige ETL-pijplijn, voeren die uit en visualiseren deze met de Dagster webserver. Net als het Prefect-dashboard biedt de Dagster webserver centrale manieren om meerdere workflows te monitoren en runs en assets te plannen.

We beginnen met het installeren van het Python-pakket.

$ pip install dagster -q

Vervolgens maken we drie Python-functies voor het extraheren, transformeren en laden van data. Deze functies heten create_dirty_data(),  clean_data(), en load_cleaned_data() in de code. Met de @asset-decorator declareren we de functies als data-assets in Dagster.

Daarna maken we de asset-job (de variabele job) met alle assets (de variabele all_assets) en maken we de assetdefinitie (de variabele defs). 

Je kunt de assetdefinitie overslaan, maar die wordt belangrijk als je je run wilt plannen, meerdere jobs wilt draaien en sensors wilt instellen.

import pandas as pd
import numpy as np
from dagster import asset, Definitions, define_asset_job, materialize

@asset
def create_dirty_data():
    # Create a sample DataFrame with dirty data
    data = {
        'Name': [' John Doe ', 'Jane Smith', 'Bob Johnson ', '  Alice Brown'],
        'Age': [30, np.nan, 40, 35],
        'City': ['New York', 'los angeles', 'CHICAGO', 'Houston'],
        'Salary': ['50,000', '60000', '75,000', 'invalid']
    }
    df = pd.DataFrame(data)
    
    # Save the DataFrame to a CSV file
    dirty_file_path = 'dag_data/dirty_data.csv'
    df.to_csv(dirty_file_path, index=False)
    
    return dirty_file_path

@asset
def clean_data(create_dirty_data):
    # Read the dirty CSV file
    df = pd.read_csv(create_dirty_data)
    
    # Clean the data
    df['Name'] = df['Name'].str.strip()
    df['Age'] = pd.to_numeric(df['Age'], errors='coerce').fillna(df['Age'].mean())
    df['City'] = df['City'].str.upper()
    df['Salary'] = df['Salary'].replace('[\$,]', '', regex=True)
    df['Salary'] = pd.to_numeric(df['Salary'], errors='coerce').fillna(0)
    
    # Calculate average salary
    avg_salary = df['Salary'].mean()
    
    # Save the cleaned DataFrame to a new CSV file
    cleaned_file_path = 'dag_data/cleaned_data.csv'
    df.to_csv(cleaned_file_path, index=False)
    
    return {
        'cleaned_file_path': cleaned_file_path,
        'avg_salary': avg_salary
    }

@asset
def load_cleaned_data(clean_data):
    cleaned_file_path = clean_data['cleaned_file_path']
    avg_salary = clean_data['avg_salary']
    
    # Read the cleaned CSV file to verify
    df = pd.read_csv(cleaned_file_path)
    
    print({
        'num_rows': len(df),
        'num_columns': len(df.columns),
        'avg_salary': avg_salary
    })

# Define all assets
all_assets = [create_dirty_data, clean_data, load_cleaned_data]

# Create a job that will materialize all assets
job = define_asset_job("all_assets_job", selection=all_assets)

# Create Definitions object
defs = Definitions(
    assets=all_assets,
    jobs=[job]
)

if __name__ == "__main__":
    result = materialize(all_assets)
    print("Pipeline execution result:", result.success)

Je kunt de bovenstaande code draaien in een Jupyter Notebook of een Python-bestand maken en het uitvoeren. 

Na uitvoering van de code krijgen we een volledige log van de workflowrun. 

Dagster webserver

Om de assets en jobruns te visualiseren, moeten we de Dagster webserver installeren en draaien. De webserver laat je jobs draaien, individuele assets materialiseren en meerdere jobs tegelijk monitoren.

$ pip install dagster-webserver

Om de Dagster-server te starten gebruiken we de Daster CLI en geven we de locatie van het Python-bestand op. In dit geval heb ik het bestand dagster_pipe.py genoemd.

$ dagster dev -f dagster_pipe.py  

Het bovenstaande commando opent de webserver automatisch in je browser. Je kunt ook rechtstreeks naar het adres http://127.0.0.1:3000 in je browser gaan.

Dagster webserver-UI.

Tot nu toe hebben we alleen de job gedeployed. Om de workflow te draaien, ga je naar het tabblad “Runs” en klik je op de knop “Launch a new run”. 

De run zou succesvol afgerond moeten zijn! Klik op de ID van de run die je interesseert om de logs te bekijken.

Dagster run-logs.

3. Mage AI

Mage AI is een open-source, hybride framework voor data-orchestratie. Hybride betekent dat je de flexibiliteit van een Jupyter Notebook krijgt en de controle van modulaire code. 

Iedereen, zelfs met beperkte Python-kennis, kan datapijplijnen bouwen, draaien en monitoren. In plaats van direct een Python-bestand te schrijven en uit te voeren, maak je een Mage AI-project en start je dat in het dashboard, waar je je datapijplijnen kunt bouwen, draaien en beheren.

Vergeleken met Airflow biedt Mage AI een gebruiksvriendelijke interface en eenvoud, wat het een uitstekende keuze maakt voor wie nieuw is in data engineering. Het is ontworpen met schaalbaarheid in gedachten en kan grote hoeveelheden data en complexe pijplijnstructuren efficiënt verwerken.

Aan de slag met Mage AI

Starten met Mage AI is vrij eenvoudig. We hoeven alleen het Mage AI Python-pakket te installeren.

$ pip install mage-ai

En het Mage AI-project starten. 

$ mage start mage_ai_etl 

Het bovenstaande commando start de webserver. Zoals eerder vermeld, gebeurt al het coderen, jobs draaien en jobs monitoren via de Mage AI-UI.

Mage AI-UI.

Klik op “+ New pipeline” om je eerste ETL-pijplijn te maken. De mijne heet “simple_etl”.

Een nieuwe pijplijn maken in Mage AI.

Vervolgens vraagt de interface je een module toe te voegen om te beginnen met coderen. Selecteer de module “Data Loader” en schrijf de volgende Python-code. 

Hier declareren we een functie create_sample_csv(), de eerste stap in onze pijplijn. We gebruiken de Mage AI-@data_loader-decorator. We definiëren ook een functie test_output() die controleert of de output bestaat. Dit helpt bij het beheren van taakafhankelijkheden.

import io
import pandas as pd

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

@data_loader
def create_sample_csv() -> pd.DataFrame:
    """
    Create a sample CSV file with duplicates and missing values
    """
    csv_data = """
category,product,quantity,price
Electronics,Laptop,5,1000
Electronics,Smartphone,10,500
Clothing,T-shirt,50,20
Clothing,Jeans,30,50
Books,Novel,100,15
Books,Textbook,20,80
Electronics,Laptop,5,1000
Clothing,T-shirt,,20
Electronics,Tablet,,300
Books,Magazine,25,
"""
    return pd.read_csv(io.StringIO(csv_data.strip()))
   
@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

De data loader-block aanmaken in Mage AI.

Maak vervolgens een andere module “Transformer” en voeg de functie clean_data() toe, zoals in de onderstaande code. 

Je kunt de functie test() negeren; je hoeft alleen de hoofdtransformerfunctie clean_data() toe te voegen.

import pandas as pd

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

@transformer
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Clean and transform the data
    """
    # Remove duplicates
    df = df.drop_duplicates()
    # Fill missing values with 0
    df = df.fillna(0)
    return df

@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

Maak op dezelfde manier een module “Data Exporter” en voeg de volgende code toe. De code declareert een dataloadfunctie, export_data_to_csv(), die de getransformeerde data opslaat in een CSV-bestand. 

import pandas as pd

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

@data_exporter
def export_data_to_csv(df: pd.DataFrame) -> None:
    """
    Export the processed data to a CSV file
    """
    df.to_csv('output_data.csv', index=False)
    print("Data exported successfully to output_data.csv")

Om de pijplijn te draaien, ga je naar het tabblad “Trigger” en klik je op “Run@once”.

De pijplijn draaien in Mage AI.

Om de run-logs te bekijken, ga je naar het tabblad “Runs” en klik je op de knop “Logs” bij de recent gedraaide pijplijn.

Mage AI flow-run-logs.

4. Kedro

Kedro is een ander populair open-source framework voor data-orchestratie dat net iets anders werkt dan de andere tools. Het is gemaakt voor machinelearning-engineers en leent veel concepten uit software-engineering, toegepast op machinelearning-projecten.

Kedro is ontworpen om zeer modulair te zijn, wat betekent dat je zelfs voor het exporteren van een dataset een datacatalogus moet maken die de locatie en het type data specificeert. Zo borg je gestandaardiseerd en efficiënt datamanagement door de hele pijplijn.

Om te begrijpen hoe Kedro past in het machinelearning-ecosysteem, kun je verschillende MLOps-tools verkennen in het artikel 25 Top MLOps Tools You Need to Know in 2024.

Vergeleken met Airflow is de Kedro-API eenvoudiger om een datapijplijn te bouwen. Het richt zich meer op machinelearning-engineering en biedt datacategorisatie en versiebeheer.

Aan de slag met Kedro

Een Kedro-datapijplijn bouwen is andere koek. Het framework is modulair, en je moet de projectstructuur en de verschillende stappen begrijpen om de workflow succesvol uit te voeren. 

Begin met het installeren van het Kedro Python-pakket. 

$ pip install kedro

Initialiseer het Kedro-project. 

$ kedro new --name=kedro_etl --tools=none --example=n 

Ga naar de projectmap. 

$ cd kedro-etl  

Maak een map in de map pipelines met de naam data_processing.

$ mkdir -p src/kedro_etl/pipelines/data_processing  

Maak een Python-bestand met de naam kedro_pipe.py en open dit in je favoriete IDE, bijvoorbeeld Visual Studio Code.

$ code src/kedro_etl/pipelines/data_processing/kedro_pipe.py

Het Python-script moet de extract-, transform- en load-functies bevatten, die knooppunten (nodes) in de pijplijn zijn. In dit geval zijn dat de functies create_sample_data(), clean_data(), en load_and_process_data().

Daarna verbinden we deze nodes met de Kedro-klasse Pipeline binnen de functie create_pipeline(). In de pijplijnfunctie definiëren we nodes en elke node heeft inputs, outputs en een node-name. 

import pandas as pd
import numpy as np
from kedro.pipeline import Pipeline, node

def create_sample_data():
    data = {
        'id': range(1, 101),
        'name': [f'Person_{i}' for i in range(1, 101)],
        'age': np.random.randint(18, 80, 100),
        'salary': np.random.randint(20000, 100000, 100),
        'missing_values': [np.nan if i % 10 == 0 else i for i in range(100)]
    }
    return pd.DataFrame(data)

def clean_data(df: pd.DataFrame):
    # Remove rows with missing values
    df_cleaned = df.dropna()

    # Convert salary to thousands
    df_cleaned['salary'] = df_cleaned['salary'] / 1000

    # Capitalize names
    df_cleaned['name'] = df_cleaned['name'].str.upper()

    return df_cleaned

def load_and_process_data(df: pd.DataFrame):
    # Calculate average salary
    avg_salary = df['salary'].mean()

    # Add a new column for salary category
    df['salary_category'] = df['salary'].apply(
        lambda x: 'High' if x > avg_salary else 'Low')

    # Calculate age groups
    df['age_group'] = pd.cut(df['age'], bins=[0, 30, 50, 100], labels=[
                             'Young', 'Middle', 'Senior'])

    print(df)
    return df

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                func=create_sample_data,
                inputs=None,
                outputs="raw_data",
                name="create_sample_data_node",
            ),
            node(
                func=clean_data,
                inputs="raw_data",
                outputs="cleaned_data",
                name="clean_data_node",
            ),
            node(
                func=load_and_process_data,
                inputs="cleaned_data",
                outputs="processed_data",
                name="load_and_process_data_node",
            ),
        ]
    )

Als we de pijplijn draaien zonder een datacatalogus te maken, wordt onze data niet geëxporteerd. We moeten dus naar het bestand conf/base/catalog.yml gaan en dit aanpassen met de datasetconfiguratie.

raw_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/sample_data.csv

cleaned_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/cleaned_data.csv

processed_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/processed_data.csv

We moeten ons nieuw gemaakte Python-bestand ook opnemen in het pipeline-register. Ga daarvoor naar het Python-bestand src/simple_etl/pipeline_registry.py en voeg de volgende code toe. 

"""Project pipelines."""
from __future__ import annotations

from kedro.pipeline import Pipeline
from kedro_etl.pipelines.data_processing import kedro_pipe

def register_pipelines() -> Dict[str, Pipeline]:
    data_processing_pipeline = kedro_pipe.create_pipeline()

    return {
        "__default__": data_processing_pipeline,
        "data_processing": data_processing_pipeline,
    }

Draai de pijplijn en bekijk de live logs in de terminal met het volgende commando.

$ kedro run

Logs van een Kedro-pijplijnrun.

Na het draaien van de pijplijn worden je bestanden als CSV opgeslagen op de locatie die in de datacatalogus is gedefinieerd.

Uitvoerbestanden van Kedro-pijplijnrun.

Als je problemen ondervindt bij het draaien van de pijplijn, overweeg dan Kedro te installeren met alle extensies. 

$ pip install "kedro[all]"

Kedro-visualisatie

We kunnen onze pijplijnen visualiseren en delen door de tool kedro-viz te installeren. 

$ pip install kedro-viz

Met het volgende commando kunnen we alle datapijplijnen en datanodes visualiseren. Het biedt ook een optie voor experimenttracering en de mogelijkheid om de pijplijnvisualisatie te delen.

$ kedro viz run

Visualisatie van een Kedro-pijplijn.

5. Luigi

Luigi is een open-source, Python-gebaseerd framework ontwikkeld door Spotify, dat uitblinkt in het beheren van langlopende batchprocessen en complexe datapijplijnen. Het is sterk in afhankelijkheidsoplossing, workflowmanagement, visualisatie en herstel na fouten, waardoor het een krachtige tool is voor het orkestreren van dataworkflows. 

Vergeleken met Airflow heeft Luigi een minimale API, kalenderscheduling en een loyale gebruikersbasis die je helpt met issues rond de data-orchestratiepijplijn. 

Aan de slag met Luigi

Voor het maken van een Luigi-datapijplijn moet je objectgeoriënteerd programmeren begrijpen. Laten we beginnen met het installeren van het Luigi Python-pakket. 

$ pip install luigi

Om een eenvoudige ETL-pijplijn in Luigi te ontwikkelen, maken we onderling verbonden taken. In plaats van Python-functies als taken te maken, maken we voor elke stap in de pijplijn een Python-klasse: FetchData, ProcessData en GenerateReport. Elke klasse heeft drie functies: requires(), output() en run(). 

De functies requires() en output() verbinden de taken, en de functie run() voert de verwerkingscode uit. Uiteindelijk bouwen we de pijplijn met de laatste taak in de pijplijn. 

import luigi
import pandas as pd
import numpy as np

class FetchData(luigi.Task):
    def output(self):
        return luigi.LocalTarget('data/fetch_data.csv')
    
    def run(self):
        # Simulate fetching data by creating a sample CSV file
        data = {
            'column1': [1, 2, np.nan, 4],
            'column2': ['A', 'B', 'C', np.nan]
        }
        df = pd.DataFrame(data)
        df.to_csv(self.output().path, index=False)

class ProcessData(luigi.Task):
    def requires(self):
        return FetchData()
    
    def output(self):
        return luigi.LocalTarget('data/process_data.csv')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Fill missing values
        df['column1'].fillna(df['column1'].mean(), inplace=True)
        df['column2'].fillna('B', inplace=True)
        df.to_csv(self.output().path, index=False)

class GenerateReport(luigi.Task):
    def requires(self):
        return ProcessData()
    
    def output(self):
        return luigi.LocalTarget('data/generate_report.txt')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Simple data analysis: calculate mean of column1 and value counts of column2
        mean_column1 = df['column1'].mean()
        value_counts_column2 = df['column2'].value_counts()
        
        with self.output().open('w') as out_file:
            out_file.write(f'Mean of column1: {mean_column1}\n')
            out_file.write('Value counts of column2:\n')
            out_file.write(value_counts_column2.to_string())

if __name__ == '__main__':
    luigi.build([GenerateReport()], local_scheduler=True)

Draai de bovenstaande code in de Jupyter Notebook of maak een Python-bestand en voer het uit via de terminal. 

Net als bij Luigi kun je ook leren hoe je een ETL-pijplijn bouwt met Apache Airflow. De tutorial behandelt de basis van extractie, transformatie en laden met Apache Airflow.

Luigi central planner

We moeten de Luigi central planner initialiseren om pijplijnruns te plannen of op een event te triggeren.

Start de scheduler door het volgende commando in de terminal te typen.

$ luigid  

2024-06-22 13:35:18,636 luigi[25056] INFO: logging configured by default settings
2024-06-22 13:35:18,636 luigi.scheduler[25056] INFO: No prior state file exists at /var/lib/luigi-server/state.pickle. Starting with empty state
2024-06-22 13:35:18,640 luigi.server[25056] INFO: Scheduler starting up

Om de pijplijn te draaien, open je een nieuwe terminal en typ je het volgende commando. Het Luigi-commando vereist een Python-bestandsnaam en de laatste taak die we willen uitvoeren. In dit geval is de bestandsnaam luigi_pipe.py en onze laatste Luigi-taak is GenerateReport.

$ python -m luigi --module luigi_pipe GenerateReport

Wil je de pijplijnrun en taakstatus visualiseren? Ga dan simpelweg naar http://localhost:8082 in je browser.

Luigi Central Planner webUI.

Daarmee ronden we onze walkthrough van de 5 beste alternatieven voor Airflow af! Wil je dieper duiken in een van de voorbeelden uit dit artikel, bekijk dan deze resources:

• Voor broncode en data van Prefect, Dagster en Luigi, zie de DataLab-werkruimte.
• Voor broncode en data van Mage AI en Kedro, zie de GitHub-repository.

Tot slot

In deze tutorial hebben we de beste open-source, gratis alternatieven voor Airflow besproken. We hebben ook elk data-orchestratiehulpmiddel bekeken en een eenvoudige ETL-pijplijn gebouwd en uitgevoerd. Het zien van codevoorbeelden helpt je bepalen welke het beste werkt voor jouw usecase.

Als je beginner bent, raad ik aan te starten met Prefect of Mage AI, omdat deze gebruiksvriendelijk zijn en een eenvoudige setup hebben. Zoek je meer geavanceerde tools die software-engineeringprincipes volgen, bekijk dan Dagster, Kedro en Luigi.

Na dit artikel is de volgende logische stap in je data-engineeringreis om een certificering te halen, zoals DataCamp’s Data Engineer in Python, om over andere tools te leren en een end-to-end datapijplijn te bouwen die je in productie kunt brengen.