データオーケストレーションにおけるAirflow代替トップ5（コード例付き）

詳しい比較は、 Airflow vs Prefect: どちらがあなたのデータワークフローに適しているか のブログを参照してください。 

Prefectのはじめ方

まずはPythonパッケージをインストールします。ターミナルで次のコマンドを実行してください。

$ pip install -U prefect

その後、prefect_etl.py というPythonスクリプトを作成し、次のコードを書き込みます。

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()

上のコードでは、extract_data()、transform_data()、およびload_data() の各タスク関数を定義し、etl() というフロー関数の中で直列に実行しています。これらの関数は、PrefectのPythonデコレータで作成されています。 

要するに、pandasのDataFrameを作成し、変換を施し、最後にprintで結果を表示しています。ETLパイプラインを簡単に模擬する方法です。

ワークフローを実行するには、次のコマンドでPythonスクリプトを走らせるだけです。

$ python prefect_etl.py 

ご覧のとおり、ワークフローは正常に完了しました。

Prefectのフロー実行ログ。

フローのデプロイ

ワークフローをスケジュールで実行したり、イベントでトリガーしたりできるよう、デプロイしていきます。フローをデプロイすれば、複数のワークフローを中央集約的に監視・管理できます。

フローのデプロイにはPrefect CLIを使用します。deploy では、Pythonファイル名、ファイル内のフロー関数名、デプロイ名を指定します。ここではデプロイ名を“simple_etl”とします。

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

上記を実行すると、デプロイを実行するワーカープールがない旨のメッセージが出る場合があります。ワーカープールを作成するには次のコマンドを使用します。

$ prefect worker start --pool 'datacamp'

ワーカープールが用意できたら、別のターミナルを開いてデプロイを実行します。prefect deployment run コマンドには「<flow-function-name>/<deployment-name>」を引数として渡します。以下のとおりです。

$ prefect deployment run 'etl/simple_etl

デプロイを実行すると、ワークフロー実行が開始された旨が表示されます。作成されるフロー実行には通常ランダムな名前が割り当てられ、私の環境では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>

詳細なログは、ワーカープールを起動したターミナルの画面に戻って確認してください。

Prefectのフロー実行サマリー。

フロー実行をよりわかりやすく可視化し、他のワークフローも管理するには、PrefectのWebサーバーを起動する必要があります。

$ prefect server start 

上記のコマンドを実行すると、Prefectのダッシュボードに自動で遷移するはずです。あるいは、ブラウザで直接http://127.0.0.1:4200 にアクセスしてもかまいません。

Prefect WebサーバーのUI

ダッシュボードからは、ワークフローの再実行、ログの閲覧、ワーカープールの確認、通知設定、その他の高度なオプションの選択が可能です。モダンなデータオーケストレーションに必要な機能が一通り揃っています。

Prefectで機械学習パイプラインを構築・実行する方法は、Using Prefect for Machine Learning Workflows のチュートリアルをご覧ください。

2. Dagster

Dagsterは、データパイプラインの定義・スケジューリング・監視のために設計されたオープンソースのフレームワークです。高いスケーラビリティを持ち、さまざまなデータチーム間のコラボレーションを促進します。 

Dagsterでは、デコレータを用いてPython関数としてデータ資産を定義できます。一度資産を定義すれば、スケジューリングやイベントトリガーでシームレスに実行できます。

Airflowと比べ、Dagsterはローカルでの開発・テスト・レビューが容易で、資産ベースのオーケストレーション手法を提供し、クラウド／コンテナネイティブです。

Dagsterのはじめ方

シンプルなETLパイプラインを作成して実行し、DagsterのWebサーバーで可視化します。Prefectのダッシュボード同様、DagsterのWebサーバーも複数のワークフローや実行、資産を一元的に監視・スケジュールできます。

まずはPythonパッケージをインストールします。

$ pip install dagster -q

次に、データの抽出・変換・ロードの3つのPython関数を作成します。ここではコード内でcreate_dirty_data()、clean_data()、およびload_cleaned_data() です。@asset デコレータを用いて、これらの関数をDagsterのデータ資産として宣言します。

次に、すべての資産（all_assets 変数）を用いてアセットジョブ（job 変数）を作成し、アセット定義（defs 変数）を作ります。 

アセット定義は省略もできますが、実行のスケジューリングや複数ジョブの実行、センサー設定を行いたい場合は重要になります。

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)

上記コードはJupyter Notebookで実行するか、Pythonファイルを作成して実行してください。 

コードを実行すると、ワークフロー実行の詳細なログが出力されます。 

DagsterのWebサーバー

資産やジョブ実行を可視化するには、DagsterのWebサーバーをインストールして起動します。Webサーバーからジョブの実行、個々の資産のマテリアライズ、複数ジョブの同時監視が可能です。

$ pip install dagster-webserver

Dagsterサーバーの起動にはDagster CLIを用い、Pythonファイルの場所を指定します。ここではファイル名をdagster_pipe.py としています。

$ dagster dev -f dagster_pipe.py  

このコマンドでブラウザが自動的に起動します。あるいは、ブラウザで直接http://127.0.0.1:3000 にアクセスしても構いません。

Dagster WebサーバーのUI。

現時点ではジョブをデプロイしただけです。ワークフローを実行するには「Runs」タブに移動し、「Launch a new run」ボタンをクリックしてください。 

実行は正常に完了するはずです。ログを見るには、対象の実行IDをクリックしてください。

Dagsterの実行ログ。

3. Mage AI

Mage AIは、オープンソースのハイブリッド型データオーケストレーションフレームワークです。ハイブリッドとは、Jupyter Notebookの柔軟性とモジュール式コードの制御性を兼ね備えることを意味します。 

Pythonの知識が限られていても、誰でもデータパイプラインを構築・実行・監視できます。Pythonファイルを直接書いて実行するのではなく、Mage AIのプロジェクトを作成してダッシュボードで起動し、そこでデータパイプラインを構築・実行・管理します。

Airflowと比べて、Mage AIはユーザーフレンドリーなインターフェースと扱いやすさが特長で、データエンジニアリング初心者にも適しています。スケーラビリティを念頭に設計されており、大量データや複雑なパイプライン構成にも効率的に対応できます。

Mage AIのはじめ方

Mage AIの開始はとても簡単です。まずはMage AIのPythonパッケージをインストールします。

$ pip install mage-ai

続いて、Mage AIプロジェクトを開始します。 

$ mage start mage_ai_etl 

このコマンドでWebサーバーが起動します。前述のとおり、コード編集・ジョブ実行・ジョブ監視はすべてMage AIのUI上で行います。

Mage AIのUI。

「+ New pipeline」をクリックして最初のETLパイプラインを作成します。ここでは名前を「simple_etl」としました。

Mage AIで新規パイプラインを作成。

続いて、コーディングを始めるためにモジュールの追加を求められます。「Data Loader」モジュールを選び、次のPythonコードを入力してください。 

ここでは、パイプラインの最初のステップとなるcreate_sample_csv() 関数を宣言し、Mage AIの@data_loader デコレータを使用します。出力の存在を検証するtest_output() 関数も定義し、タスク間の依存関係管理に役立てます。

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'

Mage AIでData Loaderブロックを作成。

同様に「Transformer」モジュールを作成し、以下のコードのとおりclean_data() 関数を追加します。 

なお、test() 関数は省略しても構いません。主な変換関数であるclean_data() を追加してください。

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'

同様に「Data Exporter」モジュールを作成し、次のコードを追加します。変換後データをCSVに保存するデータロード関数export_data_to_csv() を宣言しています。 

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")

パイプラインを実行するには、「Trigger」タブに移動し、「Run@once」をクリックします。

Mage AIでパイプラインを実行。

実行ログを見るには、「Runs」タブに移動し、直近に実行したパイプラインの「Logs」ボタンをクリックします。

Mage AIのフロー実行ログ。

4. Kedro

Kedroは、他のツールとは少し趣の異なる、人気のオープンソースのデータオーケストレーションフレームワークです。機械学習エンジニア向けに作られ、ソフトウェアエンジニアリングの多くの概念を取り入れて機械学習プロジェクトに応用しています。

Kedroは高いモジュール性を備えており、たとえばデータセットをエクスポートするだけでも、データの場所や種類を指定するデータカタログを作成する必要があります。これにより、パイプライン全体で標準化され効率的なデータ管理が可能になります。

Kedroが機械学習エコシステムの中でどのように位置づけられるかを理解するには、2024年に知っておくべきMLOpsツール25選 の記事も参考になります。

Airflowと比べると、KedroのAPIはデータパイプラインの構築がよりシンプルです。機械学習エンジニアリングに焦点を当て、データの分類やバージョニングも提供します。

Kedroのはじめ方

Kedroでのデータパイプライン構築は、ほかとは勝手が異なります。フレームワークがモジュール化されているため、プロジェクト構成やワークフロー実行のための各ステップを理解する必要があります。 

まずはKedroのPythonパッケージをインストールしましょう。 

$ pip install kedro

Kedroプロジェクトを初期化します。 

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

プロジェクトディレクトリに移動します。 

$ cd kedro-etl  

次に、pipelines フォルダ内にdata_processing というフォルダを作成します。

$ mkdir -p src/kedro_etl/pipelines/data_processing  

Pythonファイルkedro_pipe.py を作成し、お好みのIDE（たとえばVisual Studio Code）で開きます。

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

このPythonスクリプトには、抽出・変換・ロードの関数（パイプラインのノード）を含めます。ここではcreate_sample_data()、clean_data()、およびload_and_process_data() です。

続いて、KedroのPipeline クラスを使って、これらのノードをcreate_pipeline() 関数内でつなぎます。パイプライン関数では、各ノードにinputs、outputs、およびノードの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",
            ),
        ]
    )

データカタログを作成せずにパイプラインを実行すると、データはエクスポートされません。そのため、conf/base/catalog.yml を開き、データセットの設定を記述します。

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

さらに、新しく作成したPythonファイルをパイプラインレジストリに登録する必要があります。これを行うには、src/simple_etl/pipeline_registry.py を開き、次のコードを追加します。 

"""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,
    }

次のコマンドでパイプラインを実行し、ターミナルでライブログを確認します。

$ kedro run

Kedroパイプライン実行のログ。

実行後、データカタログで指定した場所にCSV形式のファイルが保存されます。

Kedroパイプライン実行の出力ファイル。

パイプラインの実行に問題がある場合は、Kedroを拡張機能込みでインストールすることを検討してください。 

$ pip install "kedro[all]"

Kedroの可視化

パイプラインの可視化と共有には、kedro-viz ツールをインストールします。 

$ pip install kedro-viz

次のコマンドを実行すると、すべてのデータパイプラインとデータノードを可視化できます。実験のトレースや可視化の共有も可能です。

$ kedro viz run

Kedroのパイプライン可視化。

5. Luigi

LuigiはSpotifyが開発した、長時間実行のバッチ処理や複雑なデータパイプラインの管理に優れた、Pythonベースのオープンソースフレームワークです。依存関係の解決、ワークフロー管理、可視化、障害復旧に強く、データワークフローのオーケストレーションに有用です。 

Airflowと比べると、LuigiはミニマルなAPI、カレンダーベースのスケジューリング、そしてデータオーケストレーションの課題解決を支援してくれる熱心なユーザーコミュニティが存在します。 

Luigiのはじめ方

Luigiでデータパイプラインを作るには、オブジェクト指向プログラミングの理解が求められます。まずはLuigiのPythonパッケージをインストールしましょう。 

$ pip install luigi

LuigiでシンプルなETLパイプラインを作成するにあたり、相互に接続されたタスクを用意します。Python関数をタスクとして作るのではなく、パイプラインの各ステップをクラス（FetchData、ProcessData、およびGenerateReport）として作成します。各クラスは、requires()、output()、run() の3つの関数を持ちます。 

これらのうち、requires() と output() がタスク同士をつなぎ、run() が処理コードを実行します。最後に、パイプラインの末端のタスクを用いてパイプラインをビルドします。 

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)

上記のコードはJupyter Notebookで実行するか、Pythonファイルを作成してターミナルから実行してください。 

Luigiと同様に、 Apache AirflowでETLパイプラインを構築する方法 も学べます。本チュートリアルでは、Airflowによる抽出・変換・ロードの基本を解説しています。

Luigi Central Planner

パイプラインのスケジュール実行やイベントトリガーを行うには、Luigi Central Plannerを初期化する必要があります。

スケジューラは次のコマンドで起動します。

$ 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

パイプラインを実行するには新しいターミナルを開き、次のコマンドを入力します。Luigiのコマンドには、Pythonファイル名と実行したい最後のタスクを指定します。ここではファイル名がluigi_pipe.py、最後のタスクがGenerateReport です。

$ python -m luigi --module luigi_pipe GenerateReport

パイプライン実行やタスクの状態を可視化したい場合は、ブラウザでhttp://localhost:8082 にアクセスするだけです。

Luigi Central PlannerのWeb UI。

以上で、Airflowの優れた代替5選の解説は完了です。この記事で紹介した各例をさらに深掘りしたい場合は、次のリソースを参照してください。

• Prefect、Dagster、Luigiのソースコードとデータは、DataLabのワークスペースを参照してください。
• Mage AIとKedroのソースコードとデータは、GitHubリポジトリを参照してください。

まとめ

本チュートリアルでは、オープンソースで無償のAirflow代替トップツールを取り上げました。各データオーケストレーションツールの特徴を学び、シンプルなETLパイプラインを構築・実行しました。コード例を見ることで、自身のユースケースに最適なツールを判断しやすくなるはずです。

初心者の方には、セットアップが簡単で扱いやすいPrefectかMage AIから始めることをおすすめします。一方で、より高度でソフトウェアエンジニアリングの実践に沿ったツールを求める場合は、Dagster、Kedro、Luigiの検討をおすすめします。

この記事を読んだ後の次の一歩としては、DataCampのData Engineer in Python などの認定取得に進み、他のツールも学びつつ、本番環境にデプロイできるエンドツーエンドのデータパイプラインを構築することをおすすめします。