configuring-airflow-language-sdks

言語 SDK 向けの Airflow の設定

言語 SDK のタスクを実行するには、Airflow が 2 つのことを把握している必要があります。ネイティブサブプロセスを起動する コーディネーター と、そのコーディネーターに振り分ける キュー です。仕組みはすべての言語 SDK で共通で、異なるのは各コーディネーターの classpath と kwargs だけです。このスキルでは共通の設定方法をまとめて説明し、その後にコーディネーターごとのオプションを紹介します。プラットフォームに依存しないため、オープンソース版の Airflow と Astro のようなマネージドプラットフォームで同じ設定を適用できます。

実験的機能です。 言語 SDK はプレビュー段階のため、設定キーが変更される可能性があります。

タスクコードについては、authoring-language-sdk-tasks（および言語ごとの作成スキル、例：authoring-java-sdk-tasks、authoring-go-sdk-tasks）を参照してください。成果物のビルドと配布については、言語ごとのデプロイスキル（例：deploying-java-sdk-bundles、deploying-go-sdk-bundles）を参照してください。

---

ワーカーの前提条件

• SDK が必要とする ランタイムまたは成果物 がワーカーノード上に存在する必要があります。これは、コーディネーターがタスクインスタンスごとにネイティブサブプロセスを起動するためです。要件は SDK によって異なります。コーディネーターごとのオプションを参照してください（Java SDK には JRE 17+ が必要です。Go SDK には言語ランタイムは不要です。バンドルは自己完結型のネイティブ実行ファイルですが、ワーカーの OS／アーキテクチャ向けにビルドする必要があります）。
• コンパイル済み／ネイティブの 成果物 をワーカーから利用できる必要があります。言語ごとのデプロイスキルを参照してください。
• コーディネーターは Airflow Task SDK（Airflow とともにインストールされる apache-airflow-task-sdk）に含まれています。追加の Python パッケージは不要です。

---

2 つの設定

どちらも [sdk] 設定セクションに記述し、すべての言語 SDK に適用されます。

1. coordinators — 任意に選んだコーディネーターの 名前 を、その実装（classpath）およびコンストラクターの kwargs に対応付ける JSON オブジェクトです。
2. queue_to_coordinator — タスクの キュー をコーディネーター名に対応付ける JSON オブジェクトです。

スタブに queue="..." を設定したタスクは、指定されたコーディネーターに渡され、そこでネイティブサブプロセスが起動されます。コーディネーター名は任意ですが、両方の設定で同じ文字列を指定する必要があります。キュー名は、Python の @task.stub に設定した queue= と一致する必要があります。

オプション A：airflow.cfg

[sdk]
coordinators = {
  "java-jdk17": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {"jars_root": ["/opt/airflow/jars"]}
  },
  "go": {
    "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator",
    "kwargs": {"executables_root": ["/opt/airflow/executable-bundles"]}
  }
}
queue_to_coordinator = {"java": "java-jdk17", "golang": "go"}

オプション B：環境変数

各値は有効な 1 行の JSONである必要があります。この形式は、コンテナー、.env ファイル、Docker Compose、Helm で便利です。

export AIRFLOW__SDK__COORDINATORS='{"java-jdk17": {"classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": {"jars_root": ["/opt/airflow/jars"]}}, "go": {"classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs": {"executables_root": ["/opt/airflow/executable-bundles"]}}}'
export AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"java": "java-jdk17", "golang": "go"}'

上記の例では複数のコーディネーター（言語ごとに 1 つ）を一度に登録し、それぞれに異なるキューを割り当てています。使用するものだけを登録してください。

---

コーディネーターごとのオプション

classpath と kwargs はコーディネーターごとに異なります。新しい言語 SDK が追加されたら、ここにサブセクションを追加してください。

JavaCoordinator

• classpath：airflow.sdk.coordinators.java.JavaCoordinator
• ワーカーのランタイム：JRE 17 以降（PATH 上の java、または java_executable を設定）。

パラメーター	デフォルト	説明
jars_root	(必須)	.jar ファイルを再帰的に検索するディレクトリを1つ以上指定します。文字列、または文字列／パスのリストを指定できます。クラスパスは自動的に構成されます。
java_executable	"java"	java バイナリへのパスです。デフォルトでは $PATH 上の java を使用します。
jvm_args	[]	JVM に渡す追加の引数です。例：["-Xmx1g", "-Dsome.property=value"]。
main_class	(自動検出)	エントリーポイントクラスを明示的に指定します。省略すると、コーディネーターは jars_root を検索し、マニフェストで Main-Class を宣言している JAR を探します。実行可能な JAR が複数ある場合は、明示的に設定してください。そうしないと、選択結果が一定しません。
task_startup_timeout	10.0	起動後、サブプロセスが接続するまで待機する秒数です。JVM の起動が遅い場合（制約のあるハードウェア、大きなクラスパス、初回のコールドスタート）は、この値を大きくしてください。

java.util.logging による Java のロギング。 SDK のロギング統合のうち、ビルド時の追加設定なしで使える依存関係は JPL と SLF4J のみです。Log4j 2 と JUL には追加設定が必要です。詳しくは deploying-java-sdk-bundles のロギング統合のセクションを参照してください。JUL で main() 内から AirflowJulHandler.setup() を呼び出す代わりに、ドキュメントに記載されている方法は logging.properties ファイルを使い、jvm_args 経由で指定することです。

[sdk]
coordinators = {
  "java-jdk17": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {
      "jars_root": ["/opt/airflow/jars"],
      "jvm_args": ["-Djava.util.logging.config.file=/opt/airflow/logging.properties"]
    }
  }
}

ExecutableCoordinator（Go およびその他の自己完結型実行可能 SDK）

• classpath：airflow.sdk.coordinators.executable.ExecutableCoordinator
• ワーカーのランタイム：バンドル以外は不要です。バンドルは自己完結型のネイティブ実行可能ファイル（AFBNDL01）であるため、言語ランタイムは必要ありません。ただし、ワーカーの OS／アーキテクチャ向けにビルドされている必要があります（一致しない場合は exec format error が発生します）。

パラメーター	デフォルト	説明
executables_root	(必須)	実行可能バンドル（末尾に AFBNDL01 が付加されたネイティブバイナリ）を再帰的に検索するディレクトリを1つ以上指定します。文字列、または文字列／パスのリストを指定できます。バンドルはファイル名ではなく、末尾のマジック値によって識別されます。コーディネーターは、受信した dag_id を各バンドルに埋め込まれたマニフェストと照合し、整合性ハッシュを検証してから起動します。
task_startup_timeout	10.0	起動後、サブプロセスが接続するまで待機する秒数です。バンドルの起動が遅い場合（制約のあるハードウェア、初回のコールドスタート）は、この値を大きくしてください。

(今後追加される他の言語向けのコーディネーターでは、それぞれの classpath、ランタイム、kwargs がここに記載されます。)

---

設定の確認

1. ワーカーの実行環境で、ランタイム／アーティファクトが使用可能であることを確認します。Java SDK の場合は、astro dev bash または docker compose exec ... 経由で java -version を実行します。Go SDK の場合は、パック済みバンドルが存在し、ワーカーの OS／アーキテクチャに合っていることを確認します。
2. kwargs で参照されるアーティファクトディレクトリ（例：jars_root、executables_root）に、ワーカーのファイルシステム上でアーティファクトが実際に含まれていることを確認します。
3. DAG を実行し、ネイティブタスクのログを開きます。サブプロセスの起動とタスクの出力が確認できるはずです。

トラブルシューティング

症状	考えられる原因／対処法
コーディネーターまたはキューに関するメッセージが表示され、タスクがすぐに失敗する	coordinators／queue_to_coordinator が有効な1行のJSONではないか、キュー名がスタブの queue= と一致していません。JSONを修正して再起動してください。
ランタイムが見つからない（例：java: command not found）	言語ランタイムがワーカーにインストールされていないか、実行可能ファイルのパスを指定するキーワード引数が間違っています。ランタイムをインストールし、バージョンを確認してください。
「アーティファクトが見つかりません」／「DAGがありません」／「dag_idを含むバンドルがありません」	アーティファクトディレクトリを指定するキーワード引数が間違っているか、アーティファクトがまだ存在しないか、アーティファクトの dag_id がスタブと一致していません。パスとIDを確認してください。
エントリーポイントが間違っている／曖昧である（Java）	jars_root 内に実行可能なJARが複数あります。main_class を明示的に設定してください。
Goバンドルが何も表示されずにスキップされる	有効なAFBNDL01バンドルではないか、整合性ハッシュの検証に失敗しました（ストリップ、署名、再ビルドの後に再パッケージしてください）。
Goバンドルで exec format error が発生する	ワーカーとは異なるOS／アーキテクチャ向けにビルドされています。--goos／--goarch でクロスコンパイルしてください（deploying-go-sdk-bundles を参照）。
DAG実行がネイティブタスクで停止する	task_startup_timeout を大きくしてください（例：30.0）。初回実行時はサブプロセスの起動に時間がかかる場合があります。

---

関連スキル

• authoring-language-sdk-tasks：共有のPythonスタブパターンと概念モデル。
• authoring-java-sdk-tasks：Javaタスクコードと対応するPythonスタブ。
• deploying-java-sdk-bundles：バンドルをビルドし、コーディネーターがスキャンする場所にアーティファクトを配置します。
• authoring-go-sdk-tasks：Goタスクコードと対応するPythonスタブ。
• deploying-go-sdk-bundles：Goバンドルをビルド／パッケージ化し、コーディネーターがスキャンする場所に配置します。
• deploying-airflow：Astro、Docker Compose、KubernetesでのAirflowの一般的なデプロイ方法。