AirflowによるスケーラブルなMLパイプライン
DAGベースのパイプライン、タスクの依存関係、データ取り込み → 学習 → 評価 → デプロイについて学習します。
「AirflowによるスケーラブルなMLパイプライン」はCoddyKit上の無料Learn AI with Pythonレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはLearn AI with Python学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Learn AI with Pythonコースには全4レッスンが含まれています。
オーケストレーションを使う理由
MLワークフローには、データの取り込み、特徴量の構築、トレーニング、評価、デプロイなど、多くのステップがあります。これらを手作業で実行すると壊れやすくなります。Apache Airflowは、スケジューリング、再試行、依存関係、監視を組み込み、これらをコードとしてオーケストレーションします。
DAG
Airflowでは、パイプラインをタスクのDAG（有向非巡回グラフ）としてモデル化します。非巡回とは、タスクがループ状に自分自身へ依存できないという意味です。そのため、開始から終了までの実行順序が常に明確になります。
DAGの定義
id、スケジュール、開始日時を指定してDAGを宣言します。スケジュールによって実行頻度が決まり、たとえば毎日の再トレーニングを設定できます。
from airflow import DAG
import datetime
with DAG(
dag_id="ml_pipeline",
schedule="@daily",
start_date=datetime.datetime(2024, 1, 1),
catchup=False,
) as dag:
...オペレーターはタスクです
DAGの各ノードは、オペレーターから作成されたタスクです。オペレーターごとに実行する処理の種類が異なり、Python関数、bashコマンド、SQLクエリなどを実行できます。
トレーニング用のPythonOperator
PythonOperatorはPythonの呼び出し可能オブジェクトを実行します。トレーニング関数を呼び出すトレーニングステップに適しています。
from airflow.operators.python import PythonOperator
def train_model():
# load features, fit model, save artifact
...
train = PythonOperator(
task_id="train",
python_callable=train_model,
)評価用のBashOperator
BashOperatorはシェルコマンドを実行します。評価スクリプトやCLIツールの呼び出しに便利です。
from airflow.operators.bash import BashOperator
evaluate = BashOperator(
task_id="evaluate",
bash_command="python /opt/ml/evaluate.py --model latest",
)依存関係の定義
>>演算子でタスクの順序を設定します。a >> bは、aが成功した後にbを実行するという意味です。これによりDAGが接続され、トレーニングが完了してから評価が開始されます。
train >> evaluate
# train must succeed before evaluate runsより長い依存関係チェーン
多数のタスクをチェーンして、パイプライン全体の順序を表現できます。Airflowは独立した分岐を並列に実行し、宣言されたすべての依存関係を守ります。
ingest >> features >> train >> evaluate >> deployXComによるデータの受け渡し
タスクは分離して実行されるため、あるタスクの結果を次のタスクに渡すにはどうすればよいのでしょうか。XCom（クロスコミュニケーション）を使うと、タスクがモデルのパスやメトリクスなどの小さな値をプッシュし、後続タスクがそれを取得できます。
def train_model(ti):
path = "/models/run_42.pt"
ti.xcom_push(key="model_path", value=path)
def deploy(ti):
path = ti.xcom_pull(key="model_path", task_ids="train")
# deploy the artifact at pathXComは小さなデータ向け
XComは小さなメタデータ（パス、ID、メトリクス）を対象としており、大規模なデータセットには適していません。大きなアーティファクトはオブジェクトストレージ（S3）に保存し、その場所だけをXCom経由で渡します。大きなペイロードにXComを使いすぎると、メタデータデータベースに負荷がかかります。
スケジューリングと再試行
Airflowは本番環境に必要な堅牢性をすぐに追加できます。スケジュールによって実行が自動的に開始され、失敗したタスクはバックオフを設けて再試行されます。また、UIですべてのタスクと実行の状態を確認でき、失敗時にはアラートも受け取れます。
クイックチェック
Airflowの知識を確認しましょう。
まとめ
Airflowを使ったスケーラブルなMLパイプラインについて学びました。
- スケジュールを設定したDAGでパイプラインを定義します
- PythonOperatorでトレーニングを実行し、BashOperatorで評価を実行します
>>演算子でタスクの依存関係を設定します- XComでタスク間に小さなアーティファクトを渡し、大きなデータはオブジェクトストレージに保存します
よくある質問
「AirflowによるスケーラブルなMLパイプライン」レッスンは無料ですか？
はい。「AirflowによるスケーラブルなMLパイプライン」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Learn AI with Pythonコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Learn AI with Pythonコースには全4レッスンが含まれています。
「AirflowによるスケーラブルなMLパイプライン」で何を学びますか？
DAGベースのパイプライン、タスクの依存関係、データ取り込み → 学習 → 評価 → デプロイについて学習します。 ブラウザで直接実行するハンズオンコードでLearn AI with Pythonを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
Learn AI with Pythonを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのLearn AI with Pythonは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「AirflowによるスケーラブルなMLパイプライン」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このLearn AI with Pythonレッスンでコードを書いて実行できますか？
はい。すべてのLearn AI with Pythonレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- AIシステムのアーキテクチャパターン
- AirflowによるスケーラブルなMLパイプライン
- 特徴量ストア：FeastとTecton
- AIシステムの可観測性とモニタリング