airflow

airflow

目的

Airflow は、データパイプラインをコードとして定義、スケジュール設定、監視するためのオープンソースのワークフローオーケストレーションツールです。Python を使用して、タスクの依存関係と実行フローを表す有向非巡回グラフ（DAG）を作成します。

使用する場面

ETL プロセス、バッチジョブ、依存関係を含む複雑なワークフローなど、定期的なデータタスクを扱う場合に Airflow を使用します。動的なスケジュール設定、再試行、データエンジニアリングパイプラインの監視が必要な場合に最適です。特に、Spark やデータベースなどのツールを使用する本番規模の運用に適しています。

主な機能

• タスク、依存関係、スケジュールを指定し、ワークフローを Python の DAG として定義します。
• タスクを指定した間隔（例：cron 形式）で実行するスケジューラーを内蔵しています。
• /admin/ などのエンドポイントを介して、タスクログや DAG のステータスを含むリアルタイム監視を行える Web UI を備えています。
• シェルコマンド用の BashOperator や、カスタム関数用の PythonOperator などのオペレーターを利用できます。
• データベース接続用の PostgresHook など、連携のための拡張可能なフックを備えています。
• airflow.cfg ファイルで設定できます。たとえば、ローカルテストでは [core] executor = LocalExecutor を設定します。

使用パターン

Airflow を使用するには、pip install apache-airflow でインストールし、airflow db init でデータベースを初期化します。Airflow のホームディレクトリにある dags フォルダーに DAG を定義します。競合を避けるため、必ず仮想環境を使用してください。認証では、ユーザーを分離するために $AIRFLOW_UID などの環境変数を設定します。

• パターン 1：スケジュールされた ETL では、センサーを使ってデータ入力を待機する、毎日実行される DAG を作成します。
• パターン 2：依存関係を設定してタスクを連結します。たとえば、データベースクエリが成功した後にのみ Python スクリプトを実行します。
• 例 1：毎日のバックアップ用のシンプルな DAG を定義します。

  from airflow import DAG
  from airflow.operators.bash import BashOperator
  dag = DAG('daily_backup', schedule_interval='@daily')
  task = BashOperator(task_id='backup', bash_command='mysqldump db > backup.sql', dag=dag)
  

• 例 2：Spark でデータを処理するパイプラインをスケジュールします。

  from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
  task = SparkSubmitOperator(task_id='spark_job', application='/path/to/script.py', dag=dag)
  

一般的なコマンド／API

環境をセットアップした後、コマンドラインから Airflow を実行します。必要に応じて、暗号化された接続には $AIRFLOW__CORE__FERNET_KEY を使用します。

• CLI コマンド：

  • データベースを初期化する：airflow db init --with-db-init
  • Web サーバーを起動する：airflow webserver --port 8080
  • スケジューラーを実行する：airflow scheduler --dag-id my_dag
  • DAG をトリガーする：airflow dags trigger my_dag --conf '{"key":"value"}'
  • DAG を一覧表示する：airflow dags list

• API エンドポイント（REST API 経由。airflow.cfg で [api] auth_backend = airflow.api.auth.backend.default を設定して有効化します）：

  • GET /api/v1/dags ですべての DAG を一覧表示します。$AIRFLOW_API_TOKEN として設定した API トークンによる認証が必要です。
  • POST /api/v1/dags/{dag_id}/dagRuns で DAG の実行をトリガーします。例：JSON ペイロード {"conf": {"param": "value"}} を使用します。
  • requests を使用した API 呼び出しの例：

    import requests
    response = requests.get('http://localhost:8080/api/v1/dags', headers={'Authorization': f'Bearer {os.environ["AIRFLOW_API_TOKEN"]}'})
    print(response.json())
    

連携に関する注意事項

フックやオペレーターを使って Airflow を他のツールと連携します。シークレットには、メタデータデータベースに保存される Airflow の Variables または Connections を使用します。データベース接続には、環境変数として $AIRFLOW_CONN_POSTGRES_DEFAULT を設定します（例：postgresql://user:pass@localhost/db）。

• Spark との連携：SparkSubmitOperator を使用し、オペレーターでエグゼキューターの設定を指定します。例：conf={"spark.executor.memory": "4g"}。
• AWS との連携：ファイル操作には S3Hook を使用し、環境変数として $AWS_ACCESS_KEY_ID と $AWS_SECRET_ACCESS_KEY を設定します。
• Kubernetes の場合は、airflow.cfg で [kubernetes] namespace = default を設定し、KubernetesPodOperator を使用します。

エラー処理

タスク定義で再試行を設定してエラーを処理します。例：retries=3, retry_delay=timedelta(minutes=5)。Web UI または airflow tasks logs <dag_id> <task_id> でログを確認します。DAG で on_failure_callback を使用してアラートをトリガーします。

• よくあるエラー：依存関係が原因でタスクが失敗します。データベース接続など、前提条件が満たされていることを確認して修正します。
• 推奨手順：タスクに email_on_failure=True を追加し、airflow.cfg で [smtp] smtp_host = your.smtp.server を設定します。
• 例：エラー処理を設定したタスクを定義します。

  from airflow.utils.email import send_email
  task = PythonOperator(task_id='failing_task', python_callable=my_function, on_failure_callback=lambda context: send_email('[email protected]', 'Task Failed', 'Error details'))
  

グラフの関係

• 関連：データパイプラインでのタスク実行に使う spark、分散処理との連携に使う hadoop、メタデータの保存に使うデータベースツール。
• 依存先：スケジューラーコンポーネントと、postgres や s3 などの外部フック。
• 連携先：データエンジニアリングクラスター内のオーケストレーションツール。たとえば、ETL フレームワークと組み合わせたワークフローに使用します。