この章では、Apache Airflow の各コンポーネントを包括的に紹介し、それぞれの役割と活用方法を学びます。

Apache Airflow 入門

DAG の数

はじめての DAG を実行する

Airflow DAG

シンプルな DAG を作成する

DAG と Airflow シェルの操作

Airflow コマンドを調べる

Airflow でタスクをテストする

Airflow Web インターフェース

Airflow UIの操作

Airflow UI で DAG を確認する

Dag のエラー

Airflow 入門

いよいよ Airflow DAG の実装の基礎を学びましょう。ハンズオン形式の演習を通じて、オペレーター、タスク、スケジューリングの設定とデプロイ方法を習得します。

Airflow のオペレーター

タスクをデコレートする

@task.bash を使った BashOperator の定義

タスクの実行順序を定義する

さらに @tasks を使う

DAGのスケジューリング

Python で DAG をスケジュールする

Airflowのスケジュールを読み解く

DAG 実行のトラブルシューティング

XCom を使ったタスク間のデータの受け渡し

XCom を使ったデータの受け渡し

XCom 変数

Airflow センサー

タスクの実行順序を確認する

FileSensor の作成

センサーの動作確認

Airflow での DAG の作成

この章では、センサーやエグゼキューターといった Airflow コンポーネントを活用して作業を効率化しながら、Airflow ワークフローの監視とトラブルシューティングの方法を学びます。

モニタリング、アラート、コールバック

コールバック関数の実装

SmtpNotifier の実装

ログには何が記録されている？

Jinja によるテンプレート

Jinja を使った記述

テンプレートへの置き換え

Airflow 変数

答えは Variable

Python で変数を読み取る

変数の設定に問題あり

Airflow のデバッグとトラブルシューティング

DAG のトラブルシューティング

from airflow.sdk import dag, task
#from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

from airflow.sdk import dag, task
from pendulum import datetime

@dag(
    dag_id="sample_dag",
    schedule="0 0 * * *",
    start_date=datetime(2023, 2, 20),
)
def sample_dag():

    @task()
    def sample():
        print("Running sample task...")

    sample()

sample_dag()

from airflow.sdk import dag, task
from airflow.providers.standard.sensors.filesystem import FileSensor
from pendulum import datetime

@dag(
    dag_id="execute_report",
    schedule="0 0 * * *",
    start_date=datetime(2026, 3, 20),
)
def execute_report():

    check_for_datafile = FileSensor(
        task_id="check_for_datafile",
        filepath="salesdata_ready.csv",
        mode="poke",
    )

    @task()
    def generate_report():
        print("Generating report...")

    check_for_datafile >> generate_report()

execute_report()

DAG が見つからない

Airflow ワークフローのメンテナンスと監視

学んだ内容を総まとめにしましょう。最終章では、これまでの知識を活かして、本番品質のワークフローを Airflow で構築します。

トリガーと障害対応

トリガールールの実装

リトライの追加

子 DAG のトリガー

分岐処理

@task.branch の定義

分岐のトラブルシューティング

HITL オペレーター

人間による承認の追加

ApprovalOperator タスク

本番パイプラインを作成する

販売 ETL パイプラインのセットアップ

月次分岐の追加

人間による承認ゲートの追加

おめでとうございます！

DAG のロジック制御

Apache Airflow 3.1.6 対応に更新されました。データを定期的に届けるには、スクリプトの作成、複雑な cron タスクの設定、変化し続ける要件への対応など、多くの手作業が伴います。チームで作業する場合、その管理はさらに困難になります。Apache Airflow を使えば、スケジューリング、エラーハンドリング、レポート機能をワークフローに組み込み、こうした煩雑さを解消できます。このコースでは、Apache Airflow の基礎をしっかりと習得し、本番環境での複雑なデータエンジニアリングパイプラインの実装方法を学びます。また、有向非巡回グラフ（DAG）の活用方法、データエンジニアリングワークフローの自動化、そして再現性の高いタスク実装の手法も身につけることができます。

Intermediate Python

Introduction to Shell

AirflowとPythonでスケジューリング、エラーハンドリング、レポートを追加しデータ処理を学ぶ。

Python で学ぶ Apache Airflow 入門

データエンジニアリングワークフローの実装とスケジューリングの方法について学びましょう。

Airflow 変数

Create Your Free Account