Data Agent Airflow スキルガイド - Data Management - Alibaba Cloud ドキュメントセンター

Data Agent は、Airflow DAG のオーサリング、インスタンスの管理、ランタイム操作の実行に役立つ、一連の組み込み Airflow スキルを提供します。自然言語の対話を通じて DMS Airflow タスクを記述するだけで、Data Agent が自動的に適切なスキルを選択します。このトピックでは、各スキルについて説明し、使用例を示し、メモリ機能を紹介します。

スキルの概要

Data Agent は、自動的にロードされる以下の 3 つの Airflow スキルをサポートしています。

スキル	シナリオ	典型的な成果物
`airflow-dms-authoring`	Airflow 3 DAG のオーサリング、変更、検証。サポートされるタスクタイプには、DMS SQL、ロックフリー SQL、Notebook、AnalyticDB Spark、Lindorm Spark、DTS データインジェスト、SSH が含まれます。スケジュール、依存関係、パラメーター、アラート、ランタイム設定も構成できます。	DAG ファイル、タスクの説明、スケジュールと依存関係の詳細、実行前設定のチェックリスト、検証結果。
`airflow-operation`	DMS Airflow インスタンスとワークスペースの管理。インスタンスの作成、クエリ、更新。仕様、ログディレクトリ、DAG とプラグインのディレクトリ、OSS マウントの設定、およびワークスペースコードのアップロードとパブリッシュ。	インスタンスの詳細、デプロイとパブリッシュのステータス、Airflow ID、失敗の概要、次のステップの提案。
`airflow-rest-api`	Airflow のランタイム設定と実行ステータスの管理。接続、変数、プール、DAG ステータス、DAG 実行、タスクインスタンス、DAG の失敗のトラブルシューティングをサポートします。	設定操作の結果、DAG 実行ステータス、失敗したタスク、エラーの概要、根本原因分析、修正の提案。

使用例

Data Agent で自然言語でタスクを記述すると、Data Agent が自動的に適切なスキルを選択します。タスクを記述する際は、ワークスペース → Airflow インスタンス → 操作のように階層的にコンテンツを指定することを推奨します。

airflow-dms-authoring

自然言語タスクの説明	Data Agent の成果物
「Airflow 3 の hello world DAG を作成してください。」	最小限の実行可能な DAG ファイル、タスクの説明、検証結果。
「この SQL 文を毎日午前 2:00 に実行する DMS SQL DAG を作成してください。」	スケジュールされた SQL DAG、スケジュールの説明、実行前設定のチェックリスト。
「これら 3 つの SQL タスク (`Task A`、`Task B`、`Task C`) を直列 DAG で連結します。」	直列依存関係の DAG とタスクの役割の説明。
「次のパラメーターを持つ Notebook / Spark / DTS DAG を作成してください...」	指定されたタスクタイプの DAG、パラメーターの説明、ランタイム設定の提案。
「この DAG に失敗アラートを追加し、Airflow 3 との互換性があるか確認してください。」	更新された DAG、アラート設定の詳細、互換性チェックの結果。

airflow-operation

自然言語タスクの説明	Data Agent の成果物
「`<region>` リージョンのワークスペース `<workspace name or ID>` に、`<specification>` 仕様で、`<instance name>` という名前の Airflow インスタンスを作成します。ネットワークと OSS の詳細は次のとおりです...」	インスタンス作成結果、Airflow ID、デプロイステータス。
「ワークスペース `<ワークスペース名または ID>` のすべての Airflow インスタンスを一覧表示します。」	インスタンスリスト、インスタンスステータス、アクション可能なインスタンス ID。
"ローカルの `<DAG ディレクトリまたはファイル>` をワークスペース `<ワークスペース名または ID>` にアップロードして公開します。"	アップロードされたファイルリスト、パブリッシュステータス、DAG 解析が保留中かどうか。

airflow-rest-api

自然言語タスクの説明	Data Agent の成果物
「ワークスペース `<workspace name or ID>` の Airflow インスタンス `<Airflow instance ID or name>` に、`<connection ID>` という名前のコネクションを作成します。」	接続作成の結果。機密フィールドは表示されません。
「ワークスペース `<workspace name or ID>` の Airflow インスタンス `<Airflow instance ID or name>` 内の DAG `<DAG ID>` をトリガーします。」	DAG 実行 ID、トリガーステータス、次のステップの提案。
「ワークスペース `<workspace name or ID>` の Airflow インスタンス `<Airflow instance ID or name>` において DAG `<DAG ID>` が失敗した理由を分析する。」	失敗した実行、失敗したタスク、主要なエラー、根本原因分析、修正の提案。

Data Agent を使用した Airflow タスクの開発

Airflow タスク開発の推奨アプローチは、2 つのステップで構成されます：

Airflow コードのオーサリング：DAG ID、スケジュール間隔、タスクタイプ、タスク内容、依存関係、ランタイム設定、アラート要件を指定します。Data Agent は、DAG ファイル、タスクの説明、実行前設定のチェックリスト、検証結果を成果物として提供します。
ワークスペースへのアップロードとパブリッシュ：ワークスペース、Airflow インスタンス、ローカルの DAG ファイルまたはディレクトリ、ターゲットのパブリッシュディレクトリ、および検証実行を行うかどうかを指定します。Data Agent は、アップロード結果、パブリッシュステータス、検証実行結果を成果物として提供します。

使用例

自然言語タスクの説明	Data Agent の成果物
「DAG ID が `<DAG ID>` で、毎日 `<schedule time>` に実行される Airflow DAG を開発します。最初に `<Task A>` を実行し、次に `<Task B>` を実行します。」	DAG ファイル、タスクの依存関係の説明、実行前設定のチェックリスト。
「生成された DAG を、ワークスペース `<ワークスペース名または ID>` の Airflow インスタンス `<Airflow インスタンス ID または名前>` に公開します。」	アップロード結果、パブリッシュステータス、DAG 解析が保留中かどうか。
「`<DAG ID>` を開発し、ワークスペース `<workspace name or ID>` の Airflow インスタンス `<Airflow instance ID or name>` に公開し、検証を実行します。」	DAG ファイル、パブリッシュステータス、検証実行結果。

メモリ機能による頻繁に使用する情報のバインド

同じワークスペースまたは Airflow インスタンスを定期的に使用する場合は、Data Agent に頻繁に使用する情報を記憶させるように依頼します。その後の会話で、「デフォルトのワークスペース」、「デフォルトの Airflow インスタンス」、またはカスタムエイリアスに言及するだけで、Data Agent はバインドされた情報を自動的に使用します。

バインドを推奨する情報：

デフォルトのワークスペース名または ID。
デフォルトの Airflow インスタンス名または ID。
優先リージョン。
頻繁に使用する DAG ID と DAG のパブリッシュディレクトリ。
頻繁に使用する接続、変数、プール名。

使用例

情報バインド時の発話内容	その後の発話内容
「私のデフォルトのワークスペース ID は `8685143589400515` です。」	「デフォルトのワークスペース内のすべての Airflow インスタンスをリストアップしてください。」
「ワークスペース `<ワークスペース名または ID>` と Airflow インスタンス `<Airflow インスタンス ID または名前>` を私のデフォルトの Airflow 環境としてバインドする。」	"ローカルの `<DAG ディレクトリまたはファイル>` を自分のデフォルトの Airflow 環境に公開する"
「DAG `<DAG ID>` は私のデフォルトのデバッグ DAG です。」	「デフォルトのデバッグ DAG をトリガーして、実行結果を表示してください。」