Jason(Zhe-You) Liu
ad3bb1f676
* Replace coordinator artifact roots with task_handler_bundle_name ADR-0013 locates Lang-SDK task handlers through a named Dag bundle, so the unversioned jars_root, executables_root and bundles_root directories go away instead of living beside it. dag_bundle_name becomes task_handler_bundle_name because the native-Dag work will use the old name with a different meaning. It has not shipped in a release, so there is no shim. A coordinator reads handlers from the named bundle at its current version, or from the task's own Dag bundle at the run's version, pinned for the task either way. CoordinatorManager.from_config validates the name of every spec without constructing coordinators, so a typo fails at config load instead of on the first task. * Serve lang-SDK harness artifacts from Dag bundles The compose e2e and the k8s lang-SDK test configured coordinators with the explicit artifact roots, which no longer exist. The Java e2e needs three bundles because one bundle is one classpath, and its three coordinators each need their own dependencies. Artifacts stay mounted on the workers only, because only the workers run them; the other components need just the bundle registration for the name to resolve. The k8s init container keeps staging from S3 because the S3 download drops the Go binary's execute bit, which the coordinator requires. * Document artifact Dag bundles for language SDK coordinators The Go, Java and TypeScript guides, the bundle specs, the READMEs and the contributor guide still pointed users at the removed jars_root, executables_root and bundles_root kwargs. A Dag bundle now decides what a coordinator sees, so two of its limits reach users. For Java, every JAR in the bundle goes on one classpath, so conflicting dependency versions cannot share a bundle, and a second Main-Class in it makes the entry point ambiguous. For Go, the coordinator skips files without the execute bit, so an object-store Dag bundle that drops it leaves the coordinator with no executable to find. * Tidy names and mocks in the artifact bundle tests Review flagged a Java test fixture still named after the removed jars_root kwarg, an e2e helper without a verb, and two execute_task patches that were not autospecced like the ones beside them. * Say artifact Dag bundles are registered on every component The guides said only the workers need the artifact Dag bundle registered, which reads as permission to register it there alone. The [sdk] configuration is also loaded by the scheduler under KubernetesExecutor, and it is rejected wherever the name is missing, so the bundle belongs in the shared bundle list. * Drop em-dashes from the rewritten doc sentences Two sentences rewritten for artifact Dag bundles kept their em-dashes, which the docs style for this work avoids. * Say the Dag processor needs the language SDK setup too In Airflow 3.4 the Dag processor checks a Python Dag's stub tasks against the task handlers the language SDK artifacts register, so a deployment that follows the old worker-only advice fails that check. task_handler_bundle_name only serves those stub tasks, which the docs did not say. * Validate task_handler_bundle_name only for routed coordinators A coordinator no queue routes to is never used, so its kwargs are not checked. Add from_config tests for an accepted name, an unrouted coordinator and a spec without the kwarg. * Add newsfragment for the removed coordinator artifact roots * Mark ADR-0013 as accepted
3.1 KiB
Scala Spark example bundle
A Scala + Apache Spark bundle for the Java SDK, exercised by the java_sdk
end-to-end test. It shows a non-Java JVM language driving Spark from a
@task.stub task and routing Log4j 2 logs into Airflow via airflow-sdk-log4j2.
The scala_spark_example Dag chains three tasks, each running in its own JVM
with a local SparkSession and passing scalar results over XCom:
spark_extract- builds a DataFrame, pushes its row count.spark_transform- aggregates total revenue.spark_load- returns the persisted total.
Build
# From java-sdk/: publish the SDK to the local Maven repository first.
./gradlew publishToMavenLocal -PskipSigning=true
cd scala_spark_example
../gradlew bundle
fatJar is disabled, so build/bundle/ holds the bundle JAR plus every runtime
JAR (Spark included). Copy it into the Dag bundle named by a Java coordinator's
task_handler_bundle_name. Every JAR in that Dag bundle goes on one classpath,
so give Spark its own Dag bundle, coordinator and queue.
Running Spark under the Java SDK
Spark on Java 17 needs a set of --add-opens / --add-exports options that open
internal JDK modules (reflection, NIO, the off-heap cleaner, Kerberos, ...) to
Spark. spark-submit and the SparkSession builder inject these through Spark's
own launcher, but the Java SDK's JavaCoordinator starts the bundle JVM directly
and bypasses that launcher, so the coordinator has to pass them itself via
jvm_args:
{
"scala-jdk": {
"classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
"kwargs": {
"task_handler_bundle_name": "scala-task-handlers",
"main_class": "org.apache.airflow.example.ScalaSparkBundleBuilder",
"jvm_args": [
"-Xmx512m",
"-XX:+IgnoreUnrecognizedVMOptions",
"--add-opens=java.base/java.lang=ALL-UNNAMED",
"--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED",
"--add-opens=java.base/sun.nio.ch=ALL-UNNAMED"
]
}
}
}
The full list mirrors org.apache.spark.launcher.JavaModuleOptions for the Spark
version pinned in build.gradle. Spark may add or drop openings between releases,
so revisit it whenever you bump Spark. The end-to-end test keeps the authoritative,
complete copy in airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
(_SPARK_JAVA_MODULE_OPTIONS).