Data Management:Airflow GUI ユーザーガイド

詳細構成 (一般)

• アラート構成：グローバルアラート設定を上書きして、現在のノードの独立したアラートポリシーを定義します。

• リトライ：グローバルリトライ設定を上書きして、現在のノードの独立したリトライポリシーを定義します。

• リソースプール構成：タスクが実行される Airflow リソースプール (Pool) を指定して、タスクの同時実行を制御します。

• トリガールール：「すべてのアップストリームタスクが成功」や「少なくとも1つのアップストリームタスクが成功」など、現在のタスクの実行条件を定義します。

基本構成 (ノードタイプ別)

次のセクションでは、いくつかの一般的なノードの基本構成について説明します。

• Single-Instance SQL

  • 機能：指定されたデータベースインスタンスで単一の SQL 文を実行します。

  • 構成：インスタンスとデータベースを選択し、SQL 文を直接記述するか、ワークスペース内の既存の .sql ファイルを選択します。

  • コールバックコード (オプション)：SQL 文が正常に実行された後に実行される Python コードを指定します。主な用途は次のとおりです。

    1. データを受け渡す (XCom)：組み込みの result 変数 ( data キーを含む辞書) を使用して SQL クエリ結果を取得し、結果をダウンストリームタスクに渡します。たとえば、result['data'][0]['co'] は、結果セットの最初の行にある co という名前の列の値を取得します。

    2. 結果の検証：SQL 結果をチェックします。条件が満たされない場合、例外をスローしてタスクを失敗させることができます。

       if len(result["data"]) > 10:
           raise Exception("too many data")

• ADB Spark

  • 機能：AnalyticDB for MySQL の Spark エンジンにジョブを送信します。

  • 構成：AnalyticDB for MySQL インスタンスとリソースグループを選択し、タスクタイプ (SQL または BATCH) に基づいてジョブを構成します。

    • SQL タスク：Spark SQL 文を直接記述するか、.sql ファイルを選択します。

    • BATCH タスク：Spark ジョブ構成を JSON 形式で提供します。

      {
        "args": [
          "1000"
        ],
        "className": "org.apache.spark.examples.SparkPi",
        "comments": [
          "-- Here is just an example of SparkPi. Modify the content and run your spark program."
        ],
        "conf": {
          "spark.driver.resourceSpec": "medium",
          "spark.executor.instances": 2,
          "spark.executor.resourceSpec": "medium"
        },
        "file": "local:///tmp/spark-examples.jar",
        "name": "SparkPi"
      }

• Notebook

  • 機能：ワークスペースで Notebook ファイルを実行します。

  • 構成：

    • Notebook セッションを選択します。

    • Notebook ファイルのパスを選択します。ファイルはワークスペースで構成されている必要があり、パスはワークスペースファイルパスである必要があります。

• Lindorm Spark

  • 機能：Lindorm Spark ジョブを送信します。

  • 構成：

    • Lindorm インスタンスを選択します。

    • ジョブタイプを選択し、ジョブパラメーターを構成します。

• Lock-Free Data Change

  • 機能：指定されたインスタンスとデータベースでロックフリー方式でデータ変更 SQL 文を実行します。構成方法は Single-Instance SQL ノードと同じです。このノードは、実行結果を処理するためのコールバックコードもサポートしています。

• Conditional Branch

  • 機能：異なる条件に基づいてワークフローパスを決定します。

  • 構成：複数のブランチ条件を追加します。各条件は True または False を返す Python 式です。式は Jinja テンプレートをサポートし、Airflow 変数を参照できます。

  • Conditional Branch には複数のダウンストリームエッジがあります。エッジをドラッグアンドドロップして設定できます。

• SSH Operations

  • 機能：SSH 経由でリモートサーバーに接続し、コマンドを実行します。

  • 構成：

    1. Airflow の Connections で SSH 接続情報を事前構成し、その Connection ID を取得します。

       

    2. ノード構成で、Connection ID と実行するシェルコマンドを入力します。

• Bash Task

  • 機能：ローカル環境で Bash コマンドを実行します。

  • 構成：

    1. 実行する Bash コマンドを構成します。

    2. コマンド実行の環境変数を構成します。

• Time Interval

  • 機能：指定された期間、ワークフローを一時停止し、その後、後続のノードの実行を続行します。

• Wait for Specified Time

  • 機能：指定された時間まで待機し、その後、後続のノードの実行を続行します。ターゲット時間は、Airflow Jinja テンプレート構文を使用して動的に設定できます。

• External Task Sensor

  • 機能：外部タスクの依存関係チェックを構成します。ワークフローは、同じ Airflow インスタンス内の外部 DAG またはタスクが完了するまで待機し、その後、実行を続行します。

  • 構成：

    • アップストリーム DAG 名：現在の DAG が依存する DAG の ID です。

    • アップストリームノード名 (オプション)：現在のタスクが依存するタスクの ID です。このパラメーターを空のままにすると、ワークフローはアップストリーム DAG 全体が完了するまで待機します。

• Empty Task

  • 機能：何も操作を実行しないプレースホルダーノードです。複数のアップストリームブランチを集約したり、ワークフロー構造を明確化するための論理的な開始点または終了点として使用できます。

  • 構成：なし。

• Data Lake Integration

  • 機能：ソースデータベースからターゲットデータレイクにデータを統合します。

  • 構成：

    • ソースデータベースとターゲットデータベースを選択します。

    • データ統合タスクのその他のパラメーターを構成します。