アセットによるデータ駆動スケジューリング

Airflow によるデータパイプラインの構築

Volker Janz

Senior Developer Advocate at Astronomer

タイミングの問題

パイプライン A は API からデータを取得します
パイプライン B はそのデータをもとにダッシュボードを構築します
両方を cron で実行すると、A の完了前に B が動く可能性があります
時間バッファを追加する方法は脆弱です

タイミングの問題

アセットとは？

from airflow.sdk import Asset
sales_data = Asset("s3://bucket/sales/daily.parquet")

一意の名前で識別されるデータへの参照です
具体的なデータエンティティを表す場合、URI をアセットに関連付けられます
ファイル、データベーステーブル、その他任意のデータを対象にできます

プロデューサーとコンシューマー

タスクはアセットを更新でき、これがアセットイベントを生成します
DAG はアセットスケジュールを使ってアセット更新時にトリガーできます

アセットの流れ

プロデューサー：outlets によるシグナル送信

sales_data = Asset("s3://bucket/sales/daily.parquet")


@task(outlets=[sales_data])
def write_sales():
    # Write data to S3
    ...

成功時に、Airflow はアセットが更新されたことを記録します
プロデューサーはコンシューマーを知る必要がありません

コンシューマー：アセットでスケジューリング

sales_data = Asset("s3://bucket/sales/daily.parquet")


@dag(schedule=[sales_data])
def build_dashboard():
    ...

cron 式は不要で、アセットの参照だけを指定します
更新後、Airflow がコンシューマーを自動的にトリガーします

条件付きスケジューリング

sales = Asset("s3://bucket/sales.parquet")
inventory = Asset("s3://bucket/inventory.parquet")


# Wait for BOTH assets to update
@dag(schedule=(sales & inventory))
def full_report(): ...


# Trigger when ANY asset updates
@dag(schedule=(sales | inventory))
def quick_refresh(): ...

CLI によるアセットの確認

$ airflow assets list

name                                  | group    | uri
s3://data-lake/sales/daily.csv        | asset    | s3://data-lake/sales/daily.csv

$ airflow assets details --name "s3://data-lake/sales/daily.csv"

property_name    | property_value
=================+=====================                                         
name             | s3://data-lake/sales/daily.csv
                ...
updated_at       | 2026-04-14T10:00:50.820041Z

airflow assets list で登録済みアセットをすべて確認できます
airflow assets details で updated_at を含むメタデータを確認できます

練習しましょう！

Airflow によるデータパイプラインの構築