パイプラインに「なぜ」を問いかける：新しい CData API プロファイルで Apache Airflow のデータを AI エージェントに活用

データチームなら誰もが一度は受け取ったことのあるメッセージがあります。「売上レポートが更新されていないんだけど、何があったの？」これに答えるには、Apache Airflow を開き、レポートの元になっている有向非巡回グラフ（DAG）を探し、実行履歴やタスクインスタンスを一つひとつクリックして確認し、上流で何が失敗したのかを突き止め、過去にも同じことが起きていないかを調べる必要があります。必要な情報はすべて Airflow のメタデータに記録されています。ただ、そこにたどり着くまでの作業は手間のかかる手作業で、しかも最も余裕のないタイミングで発生しがちです。

こうした調査は、AI エージェントにうってつけの仕事です。ただし、エージェントが Airflow のメタデータを安全に、かつ推論しやすい形で読み取れることが前提になります。CData API Driver 向けの新しい Apache Airflow API プロファイルは、まさにそれを実現します。Airflow 3 の DAG、実行（Run）、タスクインスタンス、アセット、監査ログを読み取り専用の SQL テーブルとして公開し、クエリを実行できるあらゆるエージェント、BI（ビジネスインテリジェンス）ツール、アプリケーションから利用できるようにします。

AI エージェントにパイプラインのメタデータへのアクセスを与える理由

Airflow はデータパイプラインのオーケストレーションツールとして広く使われており、自身が行ったすべての処理の詳細な履歴を記録しています。どの DAG がいつ実行され、どれくらい時間がかかったのか、どのタスクが失敗・リトライしたのか、各実行でどのデータアセットが生成されたのか、といった情報です。この履歴があれば、データチームが日々受ける次のような質問に答えられます。

• レポートやダッシュボード、下流のジョブが更新されなかったのはなぜか？
• その失敗は今日始まったのか、それともしばらく前から続いているのか？
• リトライを繰り返しているパイプラインや、だんだん遅くなっているパイプラインはどれか？
• 古くなっているデータアセットはどれか、そしてそれを生成するパイプラインの担当者は誰か？

Airflow の UI は、1つの DAG、1回の実行を個別に確認するように作られています。REST API からはすべての情報にアクセスできますが、そこから答えを導き出すにはコードを書く必要があります。認証、ページング、そして各ステップで正しい ID を使いながら DAG から実行、タスクインスタンスへとたどっていく処理です。エージェントも REST API を呼び出すことはできますが、わかりやすく定義された少数のテーブルと、すでに使い慣れたクエリ言語があるほうが、はるかに力を発揮します。

Airflow に SQL インターフェースがあれば、この課題を解決できます。一貫した方法で質問を投げかけられ、エージェントが失敗の原因を上流へたどるために必要なリレーションも把握できます。

CData API Driver と Apache Airflow API プロファイルのご紹介

CData API Driver

CData API Driver は、API プロファイルを通じて、幅広い API やサービスへの SQL アクセスを提供します。API プロファイルとは、サービスのテーブルとカラム、認証方法、API のページング方法を定義した小さなファイルです。ドライバーの Profile プロパティにプロファイルを指定し、接続プロパティを設定するだけで使い始められます。あとはドライバーが認証、ページング、クエリのプッシュダウンを処理し、JOIN、サブクエリ、集計を含む SQL-92 クエリをサポートします。

API Driver は、ほかの CData ドライバーと同じく JDBC（Java Database Connectivity）、ODBC（Open Database Connectivity）、ADO.NET、Python の各エディションで提供されており、BI や ETL（抽出・変換・ロード）向けの連携製品も用意されています。この記事では CData JDBC Driver for API を使用しました。

Airflow への接続

Apache Airflow プロファイルは Airflow 3.x とその /api/v2 REST API を対象としています（Airflow 2.x で使われていた従来の /api/v1 API はサポートしていません）。Airflow 3 では API 呼び出しを JWT（JSON Web Token）ベアラートークンで認証します。トークンは、デプロイメントの /auth/token エンドポイントにリクエストするか、Airflow の UI から生成できます。

Profile=C:\profiles\ApacheAirflow.apip;AuthScheme=APIKey;ProfileSettings='APIKey=<airflow jwt token>;Server=https://airflow.example.com';

接続で参照できる範囲は、トークンに紐づく Airflow ユーザーの権限に従います。そのため、エージェントには対象の DAG への読み取り権限を持つアカウントがあれば十分です。

データモデル

このプロファイルでは、19個のテーブルが公開されています。

調査の起点になることが多いのは、次のテーブルです。

• Dags：すべての DAG と、そのオーナー、タグ、スケジュール（TimetableSummary）、一時停止状態、次回の実行予定。
• DagRuns：各実行の状態、実行タイプ（スケジュール、手動、バックフィル、アセットトリガー）、所要時間、データ期間、実行のトリガー元（ユーザーまたはイベント）。
• TaskInstances：各タスク試行の状態、試行回数、最大試行回数、所要時間、オペレーター、プール、ホスト。
• Assets と AssetEvents：Airflow 3 のデータ駆動型スケジューリング。各アセットイベントには、どの DAG・実行・タスクがアセットを更新したのか、そしてその更新によってどの下流の実行がトリガーされたのかが記録されます。
• EventLogs：トリガー、クリア、一時停止など、ユーザーとシステムの操作の監査証跡。

たとえば次のクエリでは、タスクごとの試行回数と最大の試行番号を集計しています。

SELECT [DagId], [TaskId], COUNT(*) AS attempts, MAX([TryNumber]) AS maxTry
FROM [TaskInstances]
WHERE [DagId] IN (SELECT [DagId] FROM [Dags])
  AND [DagRunId] IN (SELECT [DagRunId] FROM [DagRuns] WHERE [DagId] IN (SELECT [DagId] FROM [Dags]))
GROUP BY [DagId], [TaskId]

実践：エージェントとダッシュボードで活用する

プロファイルの動きを実際に確認するため、オンライン小売業者を想定したモックのデータプラットフォームを用意し、Airflow 3.1 をローカルで立ち上げました。4つの DAG があり、現実でもよくある問題をいくつか仕込んでいます。

スケジューラーで19日分の履歴を生成したうえで、Airflow プロファイルを使って CData JDBC Driver for API から接続しました。

パート1：AI エージェントによる調査

AI エージェント（CData CLI 経由で動作）に与えたのは、依頼者からの次の質問だけです。

「10月5日分の日次売上レポートが出てこなかった。原因は何？過去にも起きている？どう対応すればいい？」

エージェントは経験豊富なエンジニアと同じような手順で問題を切り分け、10個のコマンド、約1分半で調査を終えました。

1. DAG の一覧（SELECT * FROM [Dags]）を読み込み、daily_revenue_report には時間ベースのスケジュールがなく、タイムテーブルが Asset になっていることに気づきました。
2. アセットをたどりました。Assets から、warehouse.orders は orders_ingest.load_orders_warehouse によって生成されて daily_revenue_report で使用されており、アセットイベントが15件あることがわかりました。
3. 実行履歴を比較しました。両方の DAG の DagRuns を見ると、orders_ingest の実行は19回（成功15回、失敗4回）で、アセットトリガーによるレポートの実行はちょうど15回。失敗した日付の実行は1件もありませんでした。
4. 失敗を掘り下げました。失敗した4回の実行の TaskInstances には、毎回同じパターンが見られました。extract_orders は成功し、validate_orders は最初で唯一の試行で失敗、load_orders_warehouse は upstream_failed としてスキップされていました。
5. 人による介入がなかったかを確認しました。EventLogs には、失敗したタスクに対するクリアやリトライの記録はありませんでした。

エージェントの回答は次のとおりです。

10月5日の日次売上レポートが実行されなかったのは、上流のデータロードが失敗したためです。daily_revenue_report はスケジュール実行ではなく、orders_ingest が warehouse.orders アセットを更新したときにトリガーされます。10月5日の orders_ingest の実行では、validate_orders が最初で唯一の試行で失敗しました（リトライは設定されていません）。そのためウェアハウスへのロードがスキップされてアセットが更新されず、レポートも開始されませんでした。

これは一度きりの問題ではありません。19回の実行のうち4回目の失敗で（9月20日、9月25日、9月30日、10月5日）、ちょうど5日おきに発生しています。偶然ではなく、データまたは検証処理に繰り返し発生する問題があると考えられます。

次の対応：10月5日の実行における validate_orders のログを確認し、データまたはチェック処理を修正したうえで、その実行の orders_ingest をクリアして再実行してください。レポートは自動的にトリガーされます。あわせて、5日周期のパターンを調査し、失敗時のアラートを追加することをお勧めします。開始されなかったレポートは、何のアラートも発しないためです。何も手を打たなければ、10月10日にも同じ問題が起きると予想されます。

エージェントは、自分には見えない情報についても明確に伝えていました。このプロファイルが公開するのは Airflow のメタデータであり、タスクログは含まれません。そのため、例外メッセージそのものは Airflow の UI で確認する必要がありますが、エージェントは開くべき DAG、実行、タスクを正確に示していました。今回のモックプラットフォームでは、ログには "Schema drift: column 'discount_code' missing from storefront extract." と記録されています。エージェントは「レポートがない」という漠然とした問い合わせを、このログ1行にまで絞り込み、さらに人間なら見落としていたかもしれない5日周期のパターンにも気づいたのです。

パート2：同じデータでパイプラインの健全性ダッシュボードを構築

エージェントが1件のインシデント調査に使うクエリは、プラットフォーム全体の健全性を把握するのにも役立ちます。そこで、同じ JDBC 接続を使って Pipeline Health ダッシュボードを構築しました。

ダッシュボードは次のパネルで構成されています。

• 主要指標：パイプライン4本、実行72回、実行成功率94.4%、失敗した実行4回、リトライが必要だった実行19回、下流のレポートが実行されなかった日4日。
• 実行カレンダー：DAG ごとに1行、日ごとに1列を割り当て、各実行を「成功」「リトライ後に成功」「失敗」「未実行」で表示します。エージェントが見つけたパターンもひと目でわかります。orders_ingest のセルが5つごとに失敗になっており、そのすぐ下の daily_revenue_report のセルは空欄です。アセットトリガーで実行されるレポートは、AssetEvents からトリガー元の取り込み実行をたどることで、カレンダー上に配置しています。
• 日別の実行時間：スケジュール実行される各 DAG の日ごとの実行時間です。marketing_attribution は期間中に6秒から22秒まで伸びています。アラートが発生することはまずない、ゆるやかな劣化の典型例です。
• 最近の失敗：失敗した各実行について、失敗したタスク、その試行回数、スキップされた下流のタスクを表示します。
• パイプライン：DAG ごとに1行で、オーナー、スケジュール、実行回数、失敗回数、リトライ回数、成功率、平均所要時間を表示します。customer_sync は19回の実行すべてでリトライが発生しており、ひときわ目立ちます。
• アセットの鮮度：warehouse.orders は19日中15日で更新されており、更新されなかった4日分が一覧表示されます。

バックエンドは、JDK（Java Development Kit）に組み込まれている HTTP サーバーを使った、Java ファイル1つだけのシンプルな構成です。ドライバー経由で Dags と AssetEvents を1回ずつ、DagRuns と TaskInstances を DAG ごとにクエリし、結果をキャッシュします。プラットフォーム全体のフルリフレッシュにかかった時間は約7.5秒でした。

使用した技術スタック

CData API Driver で Airflow のメタデータをクエリしよう

データに関する問題の多くは、その原因を Airflow がすでに把握しています。すべての実行、タスクの試行、アセットの更新がメタデータに記録されているからです。Apache Airflow API プロファイルは、このメタデータを CData API Driver 上の SQL テーブルとして利用できるようにします。これにより、パイプラインを取り巻くさまざまなツールでメタデータを活用できます。たとえば BI ダッシュボード、Airflow 自体の保持期間を超えてパイプライン履歴を保存する CData Sync ジョブ、そして一次調査を代わりに引き受けてくれる AI エージェントなどです。

今回の例では、1つの接続で2つの用途に対応できました。AI エージェントは「レポートがない」という問い合わせを、具体的な失敗タスク、繰り返し発生するパターン、そして次に取るべき対応へと落とし込みました。ダッシュボードは同じテーブルを使って、すべてのパイプラインの信頼性、リトライ、処理の遅延、アセットの鮮度を可視化しました。今回は JDBC エディションを使用しましたが、プロファイルは CData のほかのエディションでも同じように動作します。

さっそく始めましょう：CData API Driver をダウンロードして Apache Airflow プロファイルを追加し、API トークンで Airflow 3 のデプロイメントに接続したら、Dags に対して最初のクエリを実行してみてください。