authoring-java-sdk-tasks

Java SDK タスクの作成

Airflow Java SDK は、JVM 向けの言語 SDK モデルを実装しています。DAG は Python のまま保持され、各タスクインスタンスは短時間だけ実行される JVM サブプロセスで実行されます。このスキルでは、Java 固有のネイティブ API を扱います。共通モデル（Python の @task.stub パターン、ID の一致、XCom の JSON 契約）については、authoring-language-sdk-tasks に記載されています。言語 SDK を初めて使う場合は、まずそちらをお読みください。

実験的機能です。 Java SDK はプレビュー段階です。アーティファクトの座標と API は変更される場合があります。

関連スキル： authoring-language-sdk-tasks（Python スタブと共通概念）、configuring-airflow-language-sdks（キューを JavaCoordinator にルーティング）、deploying-java-sdk-bundles（JAR のコンパイルと配布）。

---

おさらい：Python 側

Java タスクには、ロジックを持たない Python スタブを対応させます。スタブではタスク、キュー、依存関係グラフ、リトライを宣言します。ID は Java のアノテーションと完全に一致させる必要があります。また、スタブの上流引数は依存関係を宣言するだけで、値は Java 側で取得されます。すべてのルールは authoring-language-sdk-tasks に記載されています。最小限の形式は次のとおりです。

from airflow.sdk import dag, task


@dag
def sales_pipeline():                     # dag_id "sales_pipeline" -> @Builder.Dag(id="sales_pipeline")
    @task.stub(queue="java")
    def extract(): ...                    # task_id "extract" -> @Builder.Task(id="extract")

    @task.stub(queue="java")
    def transform(extracted): ...

    transform(extract())


sales_pipeline()

---

Java 側：2 つの API

どちらの API も実行時の動作は同じです。スタイルに合わせて選択でき、1 つのバンドル内で組み合わせて使用することもできます。

アノテーションベースの API（推奨）

通常のクラスにアノテーションを付けると、アノテーションプロセッサーがコンパイル時に配線（<ClassName>Builder）を生成します。

import static java.lang.System.Logger.Level.INFO;
import org.apache.airflow.sdk.*;

@Builder.Dag(id = "sales_pipeline")          // must match the Python dag_id
public class SalesPipeline {
  private static final System.Logger log = System.getLogger(SalesPipeline.class.getName());

  @Builder.Task(id = "extract")              // must match the Python @task.stub name
  public long extract(Client client) {
    var conn = client.getConnection("sales_db");
    log.log(INFO, "connected to {0}", conn.host);
    return 42L;                              // return value is pushed as the return_value XCom
  }

  @Builder.Task(id = "transform")
  public long transform(
      Client client,
      @Builder.XCom(task = "extract") long recordCount) {  // pulls extract's return_value
    var threshold = (String) client.getVariable("transform_threshold");
    return recordCount * 2;
  }

  @Builder.Task   // id omitted -> the method name "load" is used
  public void load(Context context, @Builder.XCom(task = "transform") long transformed) {
    log.log(INFO, "attempt {0}, value {1}", context.ti.tryNumber, transformed);
  }
}

アノテーションのリファレンス：

アノテーション	用途
@Builder.Dag(id = "...")	クラスをタスクコンテナーとして指定します。id は Python の dag_id と一致させる必要があります。省略した場合はクラス名が使われます。省略可能な to = "..." を指定すると、生成されるビルダーの名前を変更できます（デフォルトは <ClassName>Builder）。
@Builder.Task(id = "...")	メソッドをタスクとして指定します。id は Python の @task.stub 関数名と一致させる必要があります。省略した場合はメソッド名が使われます。
@Builder.XCom(task = "...", key = "...")	上流タスクの XCom をパラメーターとして注入します。task のデフォルトはパラメーター名、key のデフォルトは生成元タスクの return_value です。パラメーター型は、保存されている JSON 値と互換性がある必要があります。

タスクメソッドの戻り値は、そのタスクの return_value XCom として自動的にプッシュされます。メソッドでは throws Exception を宣言できます。捕捉されなかった例外が発生すると、タスクインスタンスは失敗します（スタブでリトライが設定されている場合、リトライが実行されます）。

インターフェースベースの API

登録と XCom の処理を完全に制御する場合は、Task を直接実装します。

import org.apache.airflow.sdk.*;

public class ExtractTask implements Task {
  @Override
  public void execute(Context context, Client client) throws Exception {
    var conn = client.getConnection("sales_db");
    // ... do work ...
    client.setXCom(42L);   // push return_value explicitly
  }
}

タスクを手動で Dag に登録し、BundleBuilder を通じて公開します。

public class MyBundle implements BundleBuilder {
  @Override
  public Iterable<Dag> getDags() {
    var dag = new Dag("sales_pipeline");      // DAG ID matches Python
    dag.addTask("extract", ExtractTask.class);
    dag.addTask("transform", TransformTask.class);
    return java.util.List.of(dag);
  }
}

各 Task クラスには、引数なしのパブリックコンストラクターが必要です。タスク ID は DAG 内で一意でなければならず、DAG ID はバンドル内で一意でなければなりません。

---

エントリーポイント

すべてのバンドルには、DAG を SDK サーバーに渡す main があります。サーバーはコーディネーターに接続し、1 つのタスクインスタンスを実行して終了します。

import java.util.List;
import org.apache.airflow.sdk.*;

public class Main implements BundleBuilder {
  @Override
  public Iterable<Dag> getDags() {
    // With the annotation API, the *Builder classes are generated at compile time.
    return List.of(SalesPipelineBuilder.build());
  }

  public static void main(String[] args) {
    Server.create(args).serve(new Main().build());
  }
}

Server.create(args) は、Airflow がコマンドラインで渡す接続情報を解析します。自分で作成しないでください。ビルド時に、この main クラスをバンドルのメインクラスとして記録します（deploying-java-sdk-bundles を参照してください）。

---

タスクから Airflow を利用する：Client

Client はすべてのタスクに渡され、現在の DAG 実行とタスクインスタンスにスコープが限定されます。

呼び出し	戻り値	備考
client.getConnection(id)	Connection	フィールド：id、type、host、schema、login、password、port、extra。未設定のフィールドは null です。接続が存在しない場合は例外が発生します。
client.getVariable(key)	Object（または null）	期待する型にキャストします。例：(String) client.getVariable("threshold")。
client.getXCom(taskId)	Object（または null）	デフォルトでは、別のタスクの return_value を読み取ります。オーバーロードでは、DAG／実行をまたいだ読み取りやマッピングされたタスクに対応するため、key、dagId、runId、mapIndex、includePriorDates を指定できます。
client.setXCom(value)	—	return_value の XCom をプッシュします（インターフェース API）。値は JSON でシリアライズ可能である必要があります。アノテーション API では、値を返すだけで自動的に処理されます。

Context

Context パラメーターは実行メタデータを公開します：context.dagRun（dagId、runId）と context.ti（dagId、runId、taskId、mapIndex、tryNumber）。tryNumber は、リトライを考慮したロジックに役立ちます。

---

XCom：Java の型

XCom は JSON として境界を越えます（共通の契約は authoring-language-sdk-tasks にあります）。Java で読み戻すと、次の型になります。

Python の型	JSON	getXCom から返される Java の型
int	整数	Long（大きすぎる場合は BigInteger）
float	小数	Double
str	文字列	String
bool	真偽値	Boolean
None	null	null
list	配列	List<Object>
dict	オブジェクト	Map<String, Object>

@Builder.XCom のパラメーター型は、これらに一致するように宣言してください。不一致（たとえば、値が String なのに int を宣言した場合）があると、タスクは失敗します。

---

ロギング

フレームワークにかかわらず慣例的なパターンとして、クラス名と同じ名前のロガーを静的フィールドとして宣言します。

private static final System.Logger log = System.getLogger(SalesPipeline.class.getName());

レコードを Airflow のタスクログストアに届けて UI に表示するには、バンドルに SDK ロギング統合アーティファクト（airflow-sdk-jpl、airflow-sdk-slf4j、airflow-sdk-log4j2、airflow-sdk-jul のいずれか）を含める必要があります。依存関係とフレームワークごとの設定については、deploying-java-sdk-bundles のロギング統合のセクションを参照してください。airflow-sdk-jpl を使用する System.Logger（JPL）は最も軽量な選択肢で、設定も不要です。

---

SDK には完全な実例が含まれています

SDK リポジトリには、java-sdk/example/ に実行可能な例が含まれています。

• src/resources/dags/java_examples.py — Java スタブと Python タスクを組み合わせた Python DAG。retries=1 の load スタブも含まれています。
• src/java/.../AnnotationExample.java — アノテーション API。tryNumber == 1 で失敗し、リトライ時に成功するタスクも含まれています。
• src/java/.../InterfaceExampleBuilder.java — Task インターフェースと Dag.addTask(...) を使用した同じタスクです。
• src/java/.../ExampleBundleBuilder.java — 両方の DAG と main エントリーポイントを返す BundleBuilder です。

エンドツーエンドのリファレンスとして、ユーザーにはこちらを案内してください。

---

Java 固有の注意点

• Object の戻り値は適切にキャストしてください。 getVariable と getXCom は Object を返します。JSON の型に合ったキャストを使用してください（上の表を参照）。
• @Builder.XCom のパラメーター型は、保存されている JSON の型と一致している必要があります。一致しない場合、実行時にタスクが失敗します。
• アノテーション API では、アノテーションプロセッサーをビルドに含める必要があります（<ClassName>Builder を生成します）。インターフェース API では不要です。deploying-java-sdk-bundles を参照してください。
• 言語に依存しない注意点（ID の一致、タスクインスタンスごとに 1 つの JVM、スタブ上のキュー／リトライ）については、authoring-language-sdk-tasks を参照してください。

---

関連スキル

• authoring-language-sdk-tasks：Python スタブの共通パターンと概念（最初にお読みください）。
• configuring-airflow-language-sdks：java キューを JavaCoordinator にルーティングし、JRE／コーディネーターのオプションを設定します。
• deploying-java-sdk-bundles：バンドルをビルドし（Gradle／Maven）、Airflow が参照できる場所に JAR を配置します。
• authoring-dags：Airflow DAG の一般的な作成方法。