データ分析

データオーケストレーションの信頼性とパフォーマンスを向上させる Apache Airflow のスケーラブルなアラート

2024年8月27日

Christian Yarros

Strategic Cloud Engineer, Google

※この投稿は米国時間 2024 年 8 月 21 日に、Google Cloud blog に投稿されたものの抄訳です。

概要

Apache Airflow は、データワークフローのオーケストレーションツールとして人気があります。Google Cloud は、Cloud Composer と呼ばれるマネージド Airflow サービスを提供しています。これは、Apache Airflow 上に構築されたフルマネージドのワークフローオーケストレーションサービスで、パイプラインの作成や実行スケジュールの設定、パイプラインのモニタリングを可能にします。Cloud Composer を実行するときは、DAG（有向非巡回グラフ）をモニタリングし、データパイプラインのダウンタイムを最小限に抑えるために、強固なロギングとアラートを設定することが重要です。

このガイドでは、Cloud Composer のアラートの階層について説明し、Cloud Composer と Apache Airflow を使用する Google Cloud エンジニアが使用できるさまざまなアラートオプションを見ていきます。

はじめに

Cloud Composer のアラートの階層

Composer 環境

Cloud Composer 環境は、Google Kubernetes Engine に基づく自己完結型の Airflow デプロイメントであり、Airflow に組み込まれているコネクタを使用して他の Google Cloud サービスと連携します。

Cloud Composer は、ワークフローとすべての Airflow コンポーネントを実行する Google Cloud サービスをプロビジョニングします。環境の主なコンポーネントは、GKE クラスタ、Airflow ウェブサーバー、Airflow データベース、Cloud Storage バケットです。詳細については、Cloud Composer 環境のアーキテクチャをご覧ください。

このレベルで発生するアラートは主に、クラスタと Airflow コンポーネントのパフォーマンスと健全性で構成されています。

Airflow DAG Run

DAG Run は、ある時点での DAG のインスタンス化を表すオブジェクトです。DAG が実行されるたびに DAG Run が作成され、そこに含まれるすべてのタスクが実行されます。DAG Run のステータスは、タスクの状態によって異なります。各 DAG Run は互いに独立して実行されるため、ある DAG を同時に複数実行することができます。

このレベルで発生するアラートは主に、DAG Run の状態の変化（成功や失敗など）や SLA 未達で構成されています。Airflow のコールバック機能により、これらのアラートを送信するコードをトリガーできます。

Airflow Task インスタンス

Task は、Airflow における実行の基本単位です。Task は DAG に格納され、その実行順序を表すためにアップストリームとダウンストリームの依存関係が設定されます。Airflow のタスクには Operator と Sensor が含まれます。

Airflow DAG Run と同様に、Airflow Task は Airflow コールバックを介してアラートを送信するコードをトリガーできます。

まとめ

Airflow のアラート階層をまとめると、Google Cloud → Cloud Composer サービス → Cloud Composer 環境 → Airflow コンポーネント（ワーカー） → Airflow DAG Run → Airflow Task インスタンスの順になります。

Cloud Composer の本番環境レベルの実装では、この階層の各レベルでアラートとモニタリング機能を設定します。Cloud Composer には、サービス / 環境レベルでのモニタリングとアラートに関するドキュメントが幅広く用意されています。

Google Cloud での Airflow アラートの設定

次に、Airflow DAG Run と Airflow Task のレベルでアラートを発生させる 3 通りの方法を見ていきます。

オプション 1: ログベースのアラートポリシー

Google Cloud には、Airflow 環境内でのロギングとアラートに役立つネイティブツールがあります。Cloud Logging は、Airflow を含むさまざまなソースからのログを一元管理します。Cloud Monitoring を使用すると、特定のログエントリや指標のしきい値に基づいてアラートポリシーを設定できます。

含まれているログに特定のメッセージが現れるたびに通知されるようアラートポリシーを構成できます。たとえば、監査ログに特定のデータアクセスメッセージが記録されたことを知りたい場合は、そのメッセージが現れたときに通知を受け取ることができます。このようなタイプのポリシーは、ログベースのアラートポリシーと呼ばれます。詳細については、ログベースのアラートポリシーの構成 | Cloud Logging をご覧ください。

これらのサービスは、前述した Airflow のコールバック機能とうまく組み合わせることができます。手順は次のとおりです。

コールバック関数を定義し、DAG または Task レベルに設定します。
Python のネイティブロギングライブラリを使用して、特定のログメッセージを Cloud Logging に書き込みます。
特定のログメッセージによってトリガーされるログベースのアラートポリシーを定義し、通知チャンネルにアラートを送信します。

長所と短所

長所:

軽量で、設定が最小限で済む: サードパーティツール、メールサーバーの設定、追加の Airflow プロバイダは不要
ログエクスプローラやログベースの指標とのインテグレーションにより、より深い分析情報の取得や履歴分析が可能
複数の通知チャンネルオプション

短所:

メール通知アラートに最小限の情報しか含まれない
ログシンクとアラートポリシーの設定に関して習得時間とオーバーヘッドがかかる
Cloud Logging と Cloud Monitoring の使用に伴う費用

サンプルコード

Airflow DAG のコールバック:

読み込んでいます...

import time
import random
import logging
from datetime import datetime, timedelta
from airflow import models
from airflow.operators.python_operator import PythonOperator
from airflow import AirflowException

ON_DAG_FAILURE_ALERT = "Airflow DAG Failure:"
ON_SLA_MISS_ALERT = "Airflow DAG SLA Miss:"

def log_on_dag_failure(context):
    """DAG 情報を収集し、失敗時に console.log に送信する。"""
    dag = context.get('dag')

log_msg = f"""
    {ON_DAG_FAILURE_ALERT}
    *DAG*: {dag.dag_id}
    *DAG Description*: {dag.description}
    *DAG Tags*: {dag.tags}
    *Context*: {context}
    """

logging.info(log_msg)

def log_on_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis):
    """DAG 情報を収集し、SLA 未達時に console.log に送信する。"""

log_msg = f"""
        {ON_SLA_MISS_ALERT}
        *DAG*: {dag.dag_id}
        *DAG Description*: {dag.description}
        *DAG Tags*: {dag.tags}
        *Task List*: {task_list}
        *Blocking*: {blocking_task_list}
        *slas*: {slas}
        *blocking task ids*: {blocking_tis}
    """

logging.info(log_msg)

with models.DAG(
    f"log_alert_demo",
    schedule=f"*/5 * * * *", # 5 分間隔
default_args={
"owner": "Google",
"depends_on_past": False,
"retries": 1,
"retry_delay": timedelta(minutes=1),
"sla": timedelta(minutes=1), # 意図的に SLA 未達を引き起こす
},
    is_paused_upon_creation=True,
    catchup=False,
    max_active_runs=1,
    dagrun_timeout=timedelta(minutes=60),
    on_failure_callback=log_on_dag_failure,
    sla_miss_callback=log_on_sla_miss,
) as dag:

def three_minute_run():
        """
        3 分間のラン
        """
        time.sleep(180)

# 30% の確率で例外を発生させる
        if random.randint(0,9) % 3 == 0:
            raise AirflowException("Error msg")

three_minute_task = PythonOperator(
        task_id='three_minute_task', 
        python_callable=three_minute_run,
    )

three_minute_task

この Airflow DAG は、PythonOperator を使用して、定義された SLA を未達にし、Airflow 例外を発生させます。DAG Run が失敗状態になると log_on_dag_failure コールバック関数がトリガーされ、SLA が満たされなかった場合は log_on_sla_miss コールバック関数がトリガーされます。これらのコールバックはどちらも、それぞれ「Airflow DAG Failure:」と「Airflow SLA Miss:」というメッセージ文字列をログに書き込みます。これらのメッセージをログベースのアラートが捕捉し、定義された通知チャンネルにアラートを送信します。

Airflow Task のコールバック:

読み込んでいます...

import time
import random
import logging
from datetime import timedelta
from airflow import models
from airflow.operators.python_operator import PythonOperator
from airflow import AirflowException

ON_TASK_FAILURE_ALERT = "Airflow Task Failure:"

def log_on_task_failure(context):
    ti = context.get('task_instance')
    log_msg = f"""
        {ON_TASK_FAILURE_ALERT}
        *DAG*: {ti.dag_id}
        *DAG Run*: {ti.run_id}
        *Task*: {ti.task_id}
        *state*: {ti.state}
        *operator*: {ti.operator}
    """

logging.info(log_msg)

with models.DAG(
    f"log_alert_demo",
    schedule=f"*/5 * * * *", # 5 分間隔
    default_args={
        "owner": "Google",
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=1),
    },
    is_paused_upon_creation=True,
    catchup=False,
    max_active_runs=1,
    dagrun_timeout=timedelta(minutes=60),
) as dag:

def one_minute_run():
        """
        1 分間のラン
        """
        time.sleep(60)

# 50% の確率で例外を発生させる
        if random.randint(0,8) % 2 == 0:
            raise AirflowException("Error msg")

one_minute_task = PythonOperator(
        task_id='one_minute_task', 
        python_callable=one_minute_run,
        on_failure_callback=log_on_task_failure
    )

one_minute_task

この例では、タスクインスタンス自体が log_on_task_failure をコールバックします。タスクレベルで特定のコールバック関数を設定できるため、アラートを送信するタイミングと方法を特定のタスクに基づいて柔軟に決定できます。

オプション 2: SendGrid によるメール通知アラート

SendGrid は、Cloud Composer でメール通知サービスとして選ばれた SMTP サービスプロバイダです。詳細については、Cloud Composer でメール通知を構成する方法をご覧ください。

長所と短所

長所:

幅広くサポートされている信頼性の高い通知方法
分析しやすいようにフォーマットされたログスニペットを含む詳細なメール
ネイティブの Airflow EmailOperator を使用
タスクごとの柔軟な受信者リスト

短所:

大量のアラートが届いて手に負えなくなる可能性がある
外部メールプロバイダ（SendGrid）の構成とメールテンプレートの管理が必要
優先順位付けやフィルタリングを適切に行わないと、受信トレイで見失う可能性がある
SendGrid に伴う費用

サンプルコード

EmailOperator

読み込んでいます...

オプション 3: Slack や Pagerduty などのサードパーティツール

Airflow はオープンソースなので、Slack や Pagerduty といったアラートや通知を処理できる他のプロバイダも選択できます。

長所と短所

長所:

使い慣れたコミュニケーションチャンネルでのリアルタイムの通知
カスタマイズ可能なフォーマットと、特定のチャンネルやユーザーにメッセージを送信できる機能
サードパーティオプションは、チームの既存のコミュニケーションワークフローと統合されます。アラートについて直接話し合い、コンテキストや解決手順をまとめておくことができます。これにより、単独のメールやログエントリに比べて、トラブルシューティングや知識の共有が迅速化されます。

短所:

サードパーティのワークスペース、Webhook、API トークンの設定が必要
追加の Airflow 接続の管理が必要
慎重に使用しないと通知疲れにつながる可能性がある
Webhook または API トークンが侵害された場合の潜在的なセキュリティ上の懸念
サードパーティのメッセージ履歴内での長期的なログの保管が制限される可能性がある
サードパーティツールに伴う費用

サンプルコード

Slack:

読み込んでいます...

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.python import PythonOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

# 実際に使用する Slack Webhook URL に置き換える（Slack アプリ設定から取得する）
SLACK_CONN_ID = 'your_slack_connection'

def task_fail_slack_alert(context):
    slack_msg = f"""
    :red_circle: Airflow Task Failed :red_circle:
    *Task*: {context.get('task_instance').task_id}  
    *DAG*: {context.get('task_instance').dag_id}
    *Execution Time*: {context.get('execution_date')}
    *Log URL*: {context.get('task_instance').log_url}
    """
    return SlackWebhookOperator(
        task_id='slack_alert',
        http_conn_id=SLACK_CONN_ID,
        message=slack_msg,
    ).execute(context=context)  # メッセージを直ちに送信する

with DAG(
    'demo_slack_alerts',
    start_date=datetime(2024, 1, 1),
    default_args={
    "owner": "Google",
    "depends_on_past": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=1),
    "sla": timedelta(minutes=55),
     },
     is_paused_upon_creation=True,
     catchup=False,
     max_active_runs=1,
     dagrun_timeout=timedelta(minutes=60),
) as dag:

task_that_might_fail = PythonOperator(
        task_id='failing_task',
        python_callable=lambda: 1 / 0,  # ZeroDivisionError を発生させる
        on_failure_callback=task_fail_slack_alert,
    )

Pagerduty:

読み込んでいます...

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.pagerduty.notifications.pagerduty import send_pagerduty_notification

with DAG(
    "demo_pagerduty_alerts",
    start_date=datetime(2024, 1, 1),
    default_args={
    "owner": "Google",
    "depends_on_past": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=1),
    "sla": timedelta(minutes=55),
     },
     is_paused_upon_creation=True,
     catchup=False,
     max_active_runs=1,
     dagrun_timeout=timedelta(minutes=60),
     on_failure_callback=[
        send_pagerduty_notification(
            summary="The dag {{ dag.dag_id }} failed",
            severity="critical",
            source="airflow dag_id: {{dag.dag_id}}",
            dedup_key="{{dag.dag_id}}-{{ti.task_id}}",
            group="{{dag.dag_id}}",
            component="airflow",
            class_type="Prod Data Pipeline",
        )
    ],
):
    BashOperator(
        task_id="mytask",
        bash_command="fail",
        on_failure_callback=[
            send_pagerduty_notification(
                summary="The task {{ ti.task_id }} failed",
                severity="critical",
                source="airflow dag_id: {{dag.dag_id}}",
                dedup_key="{{dag.dag_id}}-{{ti.task_id}}",
                group="{{dag.dag_id}}",
                component="airflow",
                class_type="Prod Data Pipeline",
            )
        ],
    )

独自のガイダンスと次のステップ

長所と短所を考慮すると、本番環境での Airflow のアラートにはログベースのアラート ポリシー（オプション 1）をおすすめします。このアプローチをとると、スケーラブルなログ収集、シンプルなしきい値ベースのアラート、多様な通知チャンネル、指標の探索、他の Google Cloud サービスとのインテグレーションが可能になります。Logging は直感的に理解でき、Cloud Composer と統合されているため追加のプロバイダパッケージは必要ありません。

Airflow DAG にロギングとアラートを組み込むことで、データパイプラインを予防的にモニタリングし、Google Cloud の可能性を最大限に引き出すことができます。

Cloud Composer、Apache Airflow、およびこのガイドで説明されているアラートメカニズムの詳細については、次のリソースをご覧ください。

また、Cloud Composer に関連する次の Google Cloud ブログもお読みください。

ー Google、戦略的クラウド エンジニア Christian Yarros

投稿先

Data Analytics

BigQuery のベクトル検索の一般提供開始: AI を活用した新たなレベルの分析が可能に

執筆者: Joe Malone • 所要時間: 7 分

Data Analytics

Google Cloud での Apache Airflow ETL

執筆者: Jaeyeon Baek • 所要時間: 10 分

Data Analytics

BigQuery ジョブエクスプローラ: BigQuery ジョブを一元的にモニタリング、トラブルシューティング

執筆者: Vinay Yerramilli • 所要時間: 5 分

Data Analytics

現代のマーケターの戦略的アドバンテージ: AI を活用したデータクリーンルーム

執筆者: Surya Kunju • 所要時間: 8 分

データ オーケストレーションの信頼性とパフォーマンスを向上させる Apache Airflow のスケーラブルなアラート

Christian Yarros

概要

はじめに

Cloud Composer のアラートの階層

Airflow DAG Run

Airflow Task インスタンス

まとめ

Google Cloud での Airflow アラートの設定

オプション 1: ログベースのアラート ポリシー

オプション 2: SendGrid によるメール通知アラート

オプション 3: Slack や Pagerduty などのサードパーティ ツール

独自のガイダンスと次のステップ

関連記事

BigQuery のベクトル検索の一般提供開始: AI を活用した新たなレベルの分析が可能に

Google Cloud での Apache Airflow ETL

BigQuery ジョブ エクスプローラ: BigQuery ジョブを一元的にモニタリング、トラブルシューティング

現代のマーケターの戦略的アドバンテージ: AI を活用したデータ クリーンルーム

データオーケストレーションの信頼性とパフォーマンスを向上させる Apache Airflow のスケーラブルなアラート

オプション 1: ログベースのアラートポリシー

オプション 3: Slack や Pagerduty などのサードパーティツール

BigQuery ジョブエクスプローラ: BigQuery ジョブを一元的にモニタリング、トラブルシューティング

現代のマーケターの戦略的アドバンテージ: AI を活用したデータクリーンルーム