Jason(Zhe-You) Liu
52237589d0
* Java SDK: register the Scala Spark example as task handlers scala_spark_examples.py owns the Dag and declares its three tasks with @task.stub, but the bundle built a DagDef of the same ID, which is the Dag definition path. Register one handler per task ID instead, so one Dag ID is not declared on both sides. * Java SDK: use the default XCom key overloads in the Scala Spark example Client.getXCom and setXCom are @JvmOverloads, so Scala can call them with only the task ID or value and get the return_value key and current Dag. * Java SDK: shorten the Scala Spark bundle doc comment ---------
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).