Airflowプラグイン｜独自オペレーター開発と環境構築【事例付き】

記事のポイント

• カスタムオペレーターで独自の処理を追加し、ワークフローの柔軟性と再利用性を向上させ、開発時間とコストを削減。
• DockerでAirflow開発環境を構築し、公式イメージを基にカスタマイズ。プラグイン開発ではディレクトリマウントで変更を即反映。
• ログ出力とエラーハンドリングは重要。Airflowのロギング機構を利用し、try-exceptで例外処理、リトライ戦略で安定性を向上。

はじめに

Apache Airflowは、複雑なワークフローを効率的に構築、管理、監視するための強力なプラットフォームです。特に、プラグインを活用することで、Airflowの機能を拡張し、特定のニーズに合わせたカスタマイズが可能です。

本記事では、Airflowのカスタムオペレーター開発に焦点を当て、Dockerを用いた開発環境の構築から、実際のオペレーター実装、そしてトラブルシューティングまでを網羅的に解説します。

カスタムオペレーターの導入を検討している方や、Airflowの可能性をさらに引き出したいと考えている方にとって、実践的な知識と解決策を提供する一助となれば幸いです。

Airflowカスタムオペレーター開発の基礎

Airflowのカスタムオペレーター開発では、その定義と役割を理解し、プラグイン機構を通じて組み込む方法を把握することが重要です。

カスタムオペレーターのメリット

カスタムオペレーターを開発することで、既存のオペレーターでは実現できない独自の処理を追加し、ワークフローの柔軟性と再利用性を大幅に向上させることが可能です。

AirflowはPythonをベースとしているため、カスタムオペレーターはPythonで記述され、特定の業務プロセスに合わせたタスク実行を可能にします。例えば、標準のオペレーターでは対応できない特定のAPIとの連携や、独自のデータ処理ロジックを組み込むことができます。

これにより、ワークフロー全体をより効率的かつ柔軟に設計できます。また、一度開発したカスタムオペレーターは、複数のDAGで再利用できるため、開発時間とメンテナンスコストの削減にもつながります。

さらに、カスタムオペレーターは、タスクの依存関係を明確に管理し、エラーハンドリングやリカバリ処理を組み込むことで、ワークフローの安定性を高めることができます。データ分析基盤の自動化事例として、ETLプロセスをカスタムオペレーターで実装し、データ処理の効率化を図ることが挙げられます。

また、特定フォーマットへのデータ変換処理をカスタムオペレーターとして実装し、データ連携を効率化することも可能です。このように、カスタムオペレーターは、Airflowの機能を拡張し、特定のニーズに合わせたワークフローを構築するための強力なツールとなります。

カスタムオペレーター開発の注意点

カスタムオペレーターを開発する際には、Airflowのアーキテクチャとの整合性を保ち、テストとデバッグを徹底することが重要です。

Airflowの内部構造を理解せずに開発を進めると、予期せぬ問題が発生する可能性があります。例えば、オペレーターがAirflowのスケジューラーやワーカーとの連携に悪影響を与えないように注意する必要があります。

また、カスタムオペレーターは、DAGの設計思想に沿って、タスク間の依存関係を明確に定義する必要があります。テストにおいては、単体テストだけでなく、Airflow環境全体での統合テストを実施し、オペレーターが正常に動作することを確認する必要があります。

エラーハンドリングも重要な考慮事項であり、オペレーター内で発生する可能性のあるエラーを適切に処理し、ログに記録することが求められます。Docker環境で開発を行う場合は、コンテナのログを詳細に確認し、Airflow Web UIでのデバッグ機能を活用することが有効です。

さらに、リモートデバッグ環境を構築することで、より詳細なデバッグが可能になります。開発時には、Airflowのテストフレームワークを利用し、モックオブジェクトを活用することで、効率的なテストを実施できます。

• データ基盤

Airflowの使い方｜データパイプライン管理を効率化する方法 | Hakky Handbook

Apache Airflowの基本的な使い方を解説し、DAGやタスク設定の具体例を紹介します。これにより、データパイプラインの管理が効率化され、AIやデータ分析のスキル向上が期待できます。記事を読んで、キャリアの機会を広げましょう。

Dockerを用いたAirflow開発環境構築

AirflowをDockerで開発することで、環境構築が容易になり、開発から本番環境への移行もスムーズに行えます。Docker Composeを使用すると、Airflowの各コンポーネントを簡単に定義し、連携させることができます。

Dockerイメージの選定とカスタマイズ

AirflowのDockerイメージ選定では、公式が提供するイメージの利用が推奨されます。公式イメージapache/airflowは、最新バージョンの依存関係が適切に設定されており、セキュリティアップデートやバグ修正も迅速に適用されるため、安定した環境を維持できます。

例えば、Airflowのバージョン2.7.0を使用する場合は、apache/airflow:2.7.0を指定します。カスタムイメージを作成する場合は、Dockerfileを用いて、必要なライブラリや設定を追加します。FROM apache/airflow:2.7.0をベースに、COPY requirements.txt /opt/airflow/で依存関係をコピーし、RUN pip install -r /opt/airflow/requirements.txtでインストールします。これにより、Airflow環境を柔軟にカスタマイズできます。

Docker Composeの設定

Docker Composeを使用すると、Airflowの各コンポーネントをYAMLファイルで定義し、一括で起動・管理できます。docker-compose.yamlファイルには、Airflow Webサーバー、ワーカー、データベース（PostgreSQL）、Redisなどを定義します。

例えば、Webサーバーの定義では、image: apache/airflow:2.7.0でイメージを指定し、ports: - "8080:8080"でポートを公開します。環境変数の設定も重要で、AIRFLOW__CORE__SQL_ALCHEMY_CONNでデータベース接続情報を設定します。永続化ボリュームを使用すると、DAGやログなどのデータをコンテナの再起動後も保持できます。volumes: - ./dags:/opt/airflow/dagsのように設定することで、ローカルのdagsディレクトリをコンテナ内の/opt/airflow/dagsにマウントできます。

Docker環境でのプラグイン開発

Docker環境でAirflowのプラグインを開発する場合、プラグインディレクトリをコンテナにマウントすることで、開発サイクルを効率化できます。volumes: - ./plugins:/opt/airflow/pluginsのように設定すると、ローカルのpluginsディレクトリがコンテナ内の/opt/airflow/pluginsにマウントされ、プラグインの変更が即座に反映されます。

開発時には、コンテナを再起動して変更を適用する必要があります。docker-compose restart webserverコマンドを使用すると、Webサーバーコンテナを再起動できます。プラグイン開発に必要な依存関係は、Dockerfileで事前にインストールしておくことが重要です。例えば、RUN pip install --user apache-airflow[cncf.kubernetes](https://book.st-hakky.com/data-platform/plural-scaling)のように指定します。

• データ基盤

Airflow｜ローカル環境構築でハマらない！手順と注意点 | Hakky Handbook

Airflowをローカル環境で構築する手順を解説します。Dockerを使ったコンテナ化で、手軽に環境構築が可能です。この記事では、Airflowをスムーズに動かすための設定や、DAGファイルの配置方法を詳しく解説。ローカル環境でのAirflow構築に困っている方は必見です。

カスタムオペレーターの実装

カスタムオペレーターを実装することで、Airflowの機能を拡張し、特定のニーズに合わせたワークフローを構築できます。ここでは、オペレーターのパラメータ定義、外部ライブラリの利用、ログ出力とエラーハンドリングについて解説します。

オペレーターのパラメータ定義

カスタムオペレーターのパラメータ定義は、オペレーターの柔軟性と再利用性を高める上で重要です。Airflowでは、template_fieldsを活用することで、Jinjaテンプレートを使用して動的にパラメータを生成できます。これにより、実行時にパラメータを柔軟に変更することが可能になります。

例えば、日付やファイルパスなどを動的に設定できます。template_fieldsに指定されたパラメータは、Airflow UIからも入力できるようになり、ユーザーはGUIを通じて簡単にパラメータを設定できます。以下にSampleCustomOperatorの例を示します。

from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults

class SampleCustomOperator(BaseOperator):
    @apply_defaults
    def __init__(self, table_name, gcp_conn_id, execute_date, *args, **kwargs):
        super(SampleCustomOperator, self).__init__(*args, **kwargs)
        self.table_name = table_name
        self.gcp_conn_id = gcp_conn_id
        self.execute_date = execute_date

この例では、table_name、gcp_conn_id、execute_dateがパラメータとして定義されています。これらのパラメータは、apply_defaultsデコレーターによってAirflow UIから設定可能になります。

外部ライブラリの利用

カスタムオペレーターで外部ライブラリを利用することで、Airflowの機能を大幅に拡張できます。外部ライブラリを利用するためには、まず依存関係を適切に管理し、必要なライブラリをインストールする必要があります。

requirements.txtファイルにライブラリとそのバージョンを記述し、pip install -r requirements.txtコマンドでインストールします。次に、Dockerイメージにこれらのライブラリを組み込む必要があります。DockerファイルにCOPY requirements.txt /app/requirements.txtとRUN pip install -r /app/requirements.txtを追加することで、Dockerイメージの構築時にライブラリがインストールされます。

例えば、Google Cloud Platformのサービスを利用する場合は、google-cloud-bigqueryライブラリをrequirements.txtに追加し、Dockerイメージに組み込みます。これにより、BigQueryOperatorなどのカスタムオペレーターでBigQueryの操作が可能になります。

from google.cloud import bigquery

class BigQueryOperator(BaseOperator):
    @apply_defaults
    def __init__(self, gcp_conn_id, dataset_id, table_id, *args, **kwargs):
        super(BigQueryOperator, self).__init__(*args, **kwargs)
        self.gcp_conn_id = gcp_conn_id
        self.dataset_id = dataset_id
        self.table_id = table_id

    def execute(self, context):
        gcp_conn = self.hooks.get_cloud_credentials(gcp_conn_id=self.gcp_conn_id)
        client = bigquery.Client.from_service_account_json(gcp_conn)
        query_job = client.query(
            f"SELECT * FROM `{self.dataset_id}.{self.table_id}` WHERE column_name = 'value'"
        )
        results = query_job.result()

ログ出力とエラーハンドリング

カスタムオペレーターにおけるログ出力とエラーハンドリングは、ワークフローの監視とデバッグにおいて不可欠です。Airflowのロギング機構を利用することで、オペレーターの実行状況やエラーメッセージを簡単に記録できます。

loggingモジュールをインポートし、logging.info()、logging.warning()、logging.error()などのメソッドを使用してログを出力します。エラーハンドリングには、try-exceptブロックを使用し、例外が発生した場合に適切な処理を行います。リトライ戦略を実装することで、一時的なエラーから自動的に回復できます。

例えば、APIリクエストが一時的に失敗した場合に、数回リトライするように設定できます。

import logging

class LoggerOperator(BaseOperator):
    @apply_defaults
    def __init__(self, log_level=logging.INFO, *args, **kwargs):
        super(LoggerOperator, self).__init__(*args, **kwargs)
        self.log_level = log_level

    def execute(self, context):
        logging.log(self.log_level, f"Logging level: {self.log_level}")

class ErrorHandlerOperator(BaseOperator):
    @apply_defaults
    def __init__(self, exception_handling=True, *args, **kwargs):
        super(ErrorHandlerOperator, self).__init__(*args, **kwargs)
        self.exception_handling = exception_handling

    def execute(self, context):
        try:
            # 正常動作のコード
            pass
        except Exception as e:
            if self.exception_handling:
                logging.error(f"An error occurred: {str(e)}")

• データ基盤

Airflow運用術｜ログ管理、DAG設定の最適化で効率化 | Hakky Handbook

Airflowの効率的な運用方法を解説します。ログ管理、DAG設定の最適化、Deferrable Operatorの活用に焦点を当て、環境構築から運用までを網羅的にカバーします。データ分析基盤の安定稼働と効率化に不可欠な情報を提供します。Airflowを使いこなして、データ分析基盤を最適化しませんか。

▶ Hakkyのデータ基盤構築支援とは | 詳細はこちら

Airflowカスタムオペレーターのトラブルシューティング

Airflowカスタムオペレーターのトラブルシューティングでは、Dockerコンテナのログ確認、Airflow Web UIでのデバッグ、リモートデバッグ環境の構築が重要です。これらの手法を組み合わせることで、効率的な問題解決が可能になります。

Dockerコンテナのログ確認

Dockerコンテナのログを確認することで、エラーメッセージや異常な動作を把握し、問題の原因を特定できます。Airflowコンポーネント（Webサーバー、スケジューラー、ワーカー）のログ出力先は通常、/var/log/airflow/ ディレクトリです。

例えば、ワーカーのログは /var/log/airflow/worker.log に出力されます。エラーメッセージの解析では、Pythonオペレーターの例外や接続タイムアウトなどが手がかりになります。Pythonオペレーターの例外例として、ImportError: No module named 'module_name' があります。

これは、必要なモジュールがインストールされていない場合に発生します。解決策としては、requirements.txt に必要なモジュールを追加し、pip install -r requirements.txt を実行してモジュールをインストールします。

また、BigQuery Operator を使用中に Connection timed out while connecting to BigQuery というエラーが発生した場合、BigQuery への接続設定、ネットワーク、Firewall、プロキシサーバーの設定を確認する必要があります。ログを詳細に確認することで、問題の根本原因を特定し、適切な対応を取ることが可能です。

Airflow Web UIでのデバッグ

Airflow Web UIを使用すると、タスクインスタンスの詳細な実行状況やログを確認でき、デバッグに役立ちます。タスクインスタンスの詳細確認では、タスクの実行時間、試行回数、ログなどを確認できます。

これにより、タスクが失敗した原因や、パフォーマンスボトルネックを特定できます。XCom（Cross-Communication）は、タスク間でデータを共有する仕組みであり、Airflow Web UI でその値を確認できます。

例えば、あるタスクで生成されたデータが次のタスクで正しく利用されているかを確認するために、XComの値を確認します。具体的な手順としては、Airflow Web UIで対象のDAGを選択し、タスクインスタンスの詳細画面を開きます。

そこで、Logsタブを選択すると、タスクの実行ログが表示されます。また、XComタブを選択すると、タスク間で共有されたXComの値が表示されます。これらの情報を活用することで、データフローの問題やタスク間の連携の問題を効率的にデバッグできます。

リモートデバッグ環境の構築

リモートデバッグ環境を構築することで、ローカル環境からAirflowコンテナ内のコードをデバッグでき、より効率的な問題解決が可能になります。VS CodeとDockerを連携させることで、コンテナ内で実行されているPythonコードにブレークポイントを設定し、ステップ実行できます。

まず、Dockerコンテナを起動し、デバッグポートを公開します。次に、VS CodeでPython拡張機能をインストールし、リモートデバッグの設定を行います。

設定例として、launch.json ファイルに以下のような設定を追加します: { "name": "Attach to Remote", "type": "python", "request": "attach", "connect": { "host": "localhost", "port": 5678 }, "pathMappings": { "localRoot": "${workspaceFolder}", "remoteRoot": "/app" } }。

この設定により、ローカルのソースコードとコンテナ内のコードがマッピングされ、ブレークポイントが有効になります。ブレークポイントを設定し、ステップ実行を行うことで、コードの実行フローを詳細に追跡し、問題箇所を特定できます。

また、変数の値をリアルタイムで確認できるため、データの流れを把握しやすくなります。

Airflowカスタムオペレーターのテスト

Airflowカスタムオペレーターのテストは、開発したオペレーターが期待通りに動作するかを確認するために不可欠です。テストフレームワークの利用やモックオブジェクトの活用を通じて、信頼性の高いAirflowワークフローを構築できます。

Airflowのテストフレームワークの利用

Airflowのテストフレームワークを活用することで、DAGとオペレーターのテストを効率的に実行できます。pytestを基盤としたテストは、自動化されたテストプロセスを構築し、開発効率を向上させる上で重要な役割を果たします。

pytestを使用することで、ユニットテスト、インテグレーションテスト、システムテストを組み合わせることが可能です。例えば、PythonOperatorのテストでは、assert文を用いてオペレーターの状態を確認し、期待される動作を検証します。以下に、pytestを使用したテストコードの例を示します。

import pytest
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.exceptions import AirflowException
from datetime import datetime

def test_unconfigured_operator():
    with pytest.raises(AirflowException):
        dag = DAG(
            dag_id='test_unconfigured_operator',
            start_date=datetime(2023, 1, 1),
            schedule_interval=None
        )

        operator = PythonOperator(
            task_id='test_task',
            python_callable=lambda: None  # Replace with actual code or mock
        )

        assert not operator.__dict__.get('python_callable')  # Operator not configured

def test_custom_operator():
    class CustomOperator(PythonOperator):
        def __init__(self, task_id, *args, **kwargs):
            super().__init__(task_id, *args, **kwargs)

        def execute(self, context):
            return "Custom operator execution"

    dag = DAG(dag_id='test_custom_operator', start_date=datetime(2023, 1, 1))
    operator = CustomOperator(task_id='custom_task', python_callable=lambda: None)
    
    # Test if the operator executes correctly
    context = {}
    operator.execute(context=context)

この例では、pytestを使用してPythonOperatorのテストとCustomOperatorの単体テストを実行しています。assert文はオペレーターの状態を確認し、予期どおりの動作が確認されるかどうか確認しています。

DAGが複雑すぎる場合は、特定のケースでのみテスト対象のオペレーターの動作を確認するため、DAGをシンプル化することが推奨されます。ユニットテストでは、特定のステータスを確認することでオペレーターの動作が正しく動作しているかどうか確認します。

モックオブジェクトの活用

Airflowのカスタムオペレーターをテストする際、外部APIやファイルシステムなどの依存関係をモックオブジェクトで置き換えることで、テストの独立性と再現性を高めることができます。モックオブジェクトを使用することで、実際の外部システムに依存せずにテストを実行でき、テストの実行速度も向上します。

モック対象の選定基準としては、カスタムオペレータが利用する他のAirflowのコンポーネント（Hook、Sensorなど）を対象にすることが一般的です。以下に、モックオブジェクトを活用する具体的な手順を示します。

1. モックオブジェクトの作成: unittest.mockモジュールを使用して、モックオブジェクトを作成します。MagicMockクラスを使用すると、メソッドの呼び出しや属性へのアクセスを柔軟にシミュレートできます。

from unittest.mock import MagicMock

api_client_mock = MagicMock()
api_client_mock.get.return_value.json.return_value = {"status": "ok"}

2. モックオブジェクトの検証: モックオブジェクトのメソッドが正しく呼び出されたか、期待される引数で呼び出されたかを確認します。assert_called_once_withメソッドを使用すると、メソッドが特定の引数で一度だけ呼び出されたことを検証できます。

api_client_mock.get.assert_called_once_with('api_url')

3. テストデータの準備: モックオブジェクトが返す値を設定するために、テストデータを作成します。例えば、APIクライアントモックが返すJSONデータを作成するためのテストデータが必要です。

test_data = {"status": "ok"}
api_client_mock.get.return_value.json.return_value = test_data

これらの手順を踏み、モッキングされたコンポーネントを使用してカスタムオペレータをテストすることで、正確かつ迅速にテストを行うことができます。

• データ基盤

Airflow並列処理最適化｜データエンジニア向け設定術 | Hakky Handbook

Airflowの並列処理を最適化し、データパイプラインを高速化しませんか？並列処理設定を最適化することで、ジョブの実行時間を短縮し、リソースを効率的に活用できます。本記事では、並列処理を制御する主要パラメータの設定方法や最適化戦略を解説します。Airflowの並列処理を理解して、データパイプラインの効率を向上させましょう。

Airflowカスタムオペレーターのデプロイ

Airflowカスタムオペレーターを本番環境へデプロイする戦略と、プラグインの配布・インストールについて解説します。

Dockerイメージのpush

AirflowカスタムオペレーターをDockerイメージとしてpushする際、イメージレジストリの利用は不可欠です。Docker Hub、Amazon ECR、Google Container Registryなどのサービスを利用し、イメージを安全に保管・共有します。

イメージレジストリを選択する際は、セキュリティ、可用性、料金などを考慮しましょう。イメージをpushする前に、適切なタグ付けとバージョン管理を行うことが重要です。タグは、イメージのバージョンや用途を示すために使用されます。例えば、latest、stable、v1.0.0などのタグを使用できます。

バージョン管理には、Semantic Versioning（セマンティックバージョニング）を採用することを推奨します。これにより、イメージの変更履歴を追跡し、必要に応じて以前のバージョンにロールバックできます。Dockerイメージをビルドするには、docker build -t <image-name>:<tag> .コマンドを使用します。

イメージをレジストリにpushするには、まずdocker loginコマンドでレジストリにログインし、次にdocker push <image-name>:<tag>コマンドを実行します。例えば、Docker Hubにイメージをpushする場合、docker push your-username/my-airflow-operator:v1.0.0のようになります。

イメージのサイズを最適化するために、マルチステージビルドを活用しましょう。これにより、不要な依存関係やファイルを削除し、イメージサイズを削減できます。イメージのセキュリティを確保するために、定期的に脆弱性スキャンを実施しましょう。Trivyなどのツールを使用すると、イメージ内の脆弱性を自動的に検出できます。

イメージレジストリのアクセス制御を適切に設定し、許可されたユーザーのみがイメージをpull・pushできるように制限しましょう。これにより、不正なアクセスや改ざんを防ぐことができます。

Kubernetesへのデプロイ

AirflowカスタムオペレーターをKubernetesへデプロイする際、Airflow Helm Chartの利用が推奨されます。Helm Chartは、Kubernetesアプリケーションのデプロイを簡素化するパッケージマネージャーです。Airflow Helm Chartを使用すると、Airflowクラスタ全体を簡単にデプロイ・管理できます。

Helm Chartをカスタマイズすることで、カスタムプラグインをAirflow環境に組み込むことができます。カスタムプラグインを組み込むには、Helm Chartのvalues.yamlファイルを編集し、プラグインのインストールに必要な設定を追加します。例えば、カスタムプラグインのDockerイメージを指定したり、プラグインの配置場所を設定したりできます。

Kubernetes Pod Operatorを使用すると、Airflow DAGからKubernetes Podを直接操作できます。これにより、カスタムオペレーターをKubernetes上で実行し、リソースの効率的な利用やスケーラビリティの向上が期待できます。Pod Operatorを使用するには、KubernetesPodOperatorクラスをDAGに組み込みます。

Pod Operatorの設定では、Podのイメージ、リソース要件、環境変数などを指定できます。Airflowクラスタのモニタリングとロギングを適切に設定し、カスタムオペレーターの動作状況を監視しましょう。PrometheusやGrafanaなどのツールを使用すると、クラスタのパフォーマンスを可視化し、問題発生時に迅速に対応できます。ログは、ElasticsearchやFluentdなどのツールを使用して集約・分析することを推奨します。

Airflowクラスタのセキュリティを確保するために、KubernetesのRBAC（Role-Based Access Control）を適切に設定しましょう。これにより、ユーザーやサービスアカウントに適切な権限を付与し、不正なアクセスを防ぐことができます。Airflowクラスタのバックアップと復元戦略を策定し、万が一の障害に備えましょう。Veleroなどのツールを使用すると、クラスタ全体のバックアップを簡単に作成できます。

• データ基盤

Airflow権限管理｜カスタムロールで安全な開発環境を構築 | Hakky Handbook

Airflowの権限管理は、セキュアな開発に不可欠です。本記事では、カスタムロールの設計からセキュリティ設定、モニタリングまでを解説します。適切な権限設定で、情報漏洩のリスクを低減し安全な開発環境を構築しましょう。Airflowの権限管理を最適化しませんか？

おわりに

Airflowのカスタムオペレーター開発は、データパイプラインの可能性を広げ、より複雑なデータ処理ニーズに対応するための鍵となります。しかし、開発、テスト、デプロイには専門的な知識と経験が不可欠です。

もし、データ基盤構築でお困りの際は、ぜひHakkyにご相談ください。お客様のビジネスに最適なデータ基盤を構築し、データ活用を加速させるお手伝いをいたします。

お知らせ

Hakkyでは、お客様のビジネスに合わせた最適なデータ基盤構築を支援します。 Airflowカスタムオペレーターに関する知見を活かし、データ基盤構築をより効率的にしませんか。

• お問い合わせはこちら
• 資料請求はこちら

関連記事

• データ基盤

Airflowで定期実行を自動化｜データ処理の効率化を実現する方法 | Hakky Handbook

Apache Airflowを活用することで、データ処理の自動化が実現できます。定期実行設定を行うことで、業務の生産性が向上し、データ分析の効率が大幅に改善されます。詳細な設定方法や事例を通じて、Airflowの活用法を学びましょう。

参考文献

• Apache Airflowとは？徹底解説
• 【初心者向け】Airflowとは？Airflowに関してわかりやすく解説！ - AI Academy Media
• Airflowはすごいぞ！100行未満で本格的なデータパイプライン #Python - Qiita
• 機械学習におけるApache Airflowの活用方法
• Airflowとは？最新の機能と実践的な活用法を徹底解説 | Data Driven Knowledgebase