You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by ha...@apache.org on 2026/10/20 12:54:23 UTC
(hop) branch main updated: Issue #9016 : Run native Spark pipelines on YARN from Hop GUI, hop-run and Hop Server, fixes #9020 (#9021)
This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new c135c175b6 Issue #9016 : Run native Spark pipelines on YARN from Hop GUI, hop-run and Hop Server, fixes #9020 (#9021)
c135c175b6 is described below
commit c135c175b6364cc589e5c9f361f1550ffd32e2d5
Author: Bart Maertens <ba...@know.bi>
AuthorDate: Tue Oct 20 14:54:18 2026 +0200
Issue #9016 : Run native Spark pipelines on YARN from Hop GUI, hop-run and Hop Server, fixes #9020 (#9021)
---
.../integration-tests-spark-native-yarn.yaml | 65 ++++
docker/integration-tests/spark-yarn/Dockerfile | 28 ++
.../spark-yarn/conf/core-site.xml | 29 ++
.../spark-yarn/conf/hdfs-site.xml | 35 +++
.../spark-yarn/conf/yarn-site.xml | 63 ++++
docker/integration-tests/spark-yarn/start.sh | 36 +++
docs/hop-user-manual/modules/ROOT/nav.adoc | 1 +
.../native-spark-pipeline-engine.adoc | 10 +-
.../spark/getting-started-with-native-spark.adoc | 10 +-
.../pipeline/spark/local-standalone-cluster.adoc | 13 +-
.../modules/ROOT/pages/pipeline/spark/yarn.adoc | 182 ++++++++++++
.../0010-filter-text-outputs-validation.hpl | 328 +++++++++++++++++++++
.../spark-native/0010-filter-text-outputs.hpl | 207 +++++++++++++
.../spark-native/main-0010-filter-text-outputs.hwf | 139 +++++++++
.../0010-filter-text-outputs-validation UNIT.json | 57 ++++
plugins/engines/spark/pom.xml | 25 ++
.../apache/hop/spark/core/HopMapPartitionsFn.java | 15 +-
.../org/apache/hop/spark/core/HopSparkUtil.java | 8 +
.../ISparkPipelineEngineRunConfiguration.java | 7 +
.../hop/spark/engines/SparkPipelineEngine.java | 106 ++++++-
.../engines/SparkPipelineRunConfiguration.java | 10 +
.../engines/template/SparkRunConfigTemplate.java | 3 +-
.../pipeline/HopPipelineMetaToSparkConverter.java | 60 +++-
.../handler/SparkGenericTransformHandler.java | 2 +
.../java/org/apache/hop/spark/run/MainSpark.java | 18 ++
.../hop/spark/util/SparkHadoopConfiguration.java | 205 +++++++++++++
.../engines/messages/messages_en_US.properties | 2 +
.../engines/SparkPipelineRunConfigurationTest.java | 2 +
.../spark/util/SparkHadoopConfigurationTest.java | 117 ++++++++
29 files changed, 1760 insertions(+), 23 deletions(-)
diff --git a/docker/integration-tests/integration-tests-spark-native-yarn.yaml b/docker/integration-tests/integration-tests-spark-native-yarn.yaml
new file mode 100644
index 0000000000..6bd8e69623
--- /dev/null
+++ b/docker/integration-tests/integration-tests-spark-native-yarn.yaml
@@ -0,0 +1,65 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Single-node HDFS + YARN for running native Spark pipelines on YARN from Hop GUI, hop-run or Hop
+# Server on the Docker host (Linux).
+#
+# docker compose -f docker/integration-tests/integration-tests-spark-native-yarn.yaml up -d --build
+#
+# Hop side (native Spark run configuration):
+# Spark master yarn
+# Hadoop configuration folder <hop source>/docker/integration-tests/spark-yarn/conf
+# Fat jar file location a native-provided fat jar
+# Spark config spark.driver.host=172.31.250.1
+# spark.executorEnv.JAVA_HOME=/opt/java21
+# spark.yarn.appMasterEnv.JAVA_HOME=/opt/java21
+#
+# 172.31.250.1 is the Docker host on the network below: YARN containers connect back to the Spark
+# driver (Hop) there. HDFS: http://localhost:9870, YARN: http://localhost:8088
+
+services:
+ hadoop-yarn:
+ build:
+ context: ./spark-yarn
+ user: root
+ volumes:
+ - ./spark-yarn/conf/core-site.xml:/opt/hadoop/etc/hadoop/core-site.xml:ro
+ - ./spark-yarn/conf/hdfs-site.xml:/opt/hadoop/etc/hadoop/hdfs-site.xml:ro
+ - ./spark-yarn/conf/yarn-site.xml:/opt/hadoop/etc/hadoop/yarn-site.xml:ro
+ ports:
+ - "8020:8020"
+ - "8032:8032"
+ - "8088:8088"
+ - "9870:9870"
+ # The datanode registers as localhost: HDFS clients on the host write blocks here
+ - "9864:9864"
+ - "9866:9866"
+ networks:
+ - spark-yarn
+ healthcheck:
+ test: ["CMD-SHELL", "hdfs dfsadmin -safemode get | grep -q OFF && yarn node -list 2>/dev/null | grep -q RUNNING"]
+ interval: 10s
+ timeout: 20s
+ retries: 30
+ start_period: 30s
+
+networks:
+ spark-yarn:
+ ipam:
+ config:
+ - subnet: 172.31.250.0/24
+ gateway: 172.31.250.1
diff --git a/docker/integration-tests/spark-yarn/Dockerfile b/docker/integration-tests/spark-yarn/Dockerfile
new file mode 100644
index 0000000000..47bd68e278
--- /dev/null
+++ b/docker/integration-tests/spark-yarn/Dockerfile
@@ -0,0 +1,28 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Single-node HDFS + YARN for the native Spark engine.
+# Hadoop runs on the image's own Java; YARN containers (Spark AM and executors) run on the
+# Java 21 runtime in /opt/java21, which the Hop classes in the fat jar need.
+
+FROM eclipse-temurin:21-jre AS java21
+
+FROM apache/hadoop:3.4.2
+COPY --from=java21 /opt/java/openjdk /opt/java21
+COPY --chmod=755 start.sh /opt/start-hadoop-yarn.sh
+ENV HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
+CMD ["/opt/start-hadoop-yarn.sh"]
diff --git a/docker/integration-tests/spark-yarn/conf/core-site.xml b/docker/integration-tests/spark-yarn/conf/core-site.xml
new file mode 100644
index 0000000000..9d07d40207
--- /dev/null
+++ b/docker/integration-tests/spark-yarn/conf/core-site.xml
@@ -0,0 +1,29 @@
+<?xml version="1.0"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<!-- Single-node HDFS + YARN for native Spark tests. The same files serve the container and
+ the Hop client on the Docker host: everything is reached as localhost. -->
+<configuration>
+ <property>
+ <name>fs.defaultFS</name>
+ <value>hdfs://localhost:8020</value>
+ </property>
+ <property>
+ <name>hadoop.tmp.dir</name>
+ <value>/tmp/hadoop</value>
+ </property>
+</configuration>
diff --git a/docker/integration-tests/spark-yarn/conf/hdfs-site.xml b/docker/integration-tests/spark-yarn/conf/hdfs-site.xml
new file mode 100644
index 0000000000..4fd878e693
--- /dev/null
+++ b/docker/integration-tests/spark-yarn/conf/hdfs-site.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<configuration>
+ <property>
+ <name>dfs.replication</name>
+ <value>1</value>
+ </property>
+ <property>
+ <name>dfs.permissions.enabled</name>
+ <value>false</value>
+ </property>
+ <property>
+ <name>dfs.namenode.rpc-bind-host</name>
+ <value>0.0.0.0</value>
+ </property>
+ <property>
+ <name>dfs.namenode.http-bind-host</name>
+ <value>0.0.0.0</value>
+ </property>
+</configuration>
diff --git a/docker/integration-tests/spark-yarn/conf/yarn-site.xml b/docker/integration-tests/spark-yarn/conf/yarn-site.xml
new file mode 100644
index 0000000000..cdce7713bb
--- /dev/null
+++ b/docker/integration-tests/spark-yarn/conf/yarn-site.xml
@@ -0,0 +1,63 @@
+<?xml version="1.0"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<configuration>
+ <property>
+ <name>yarn.resourcemanager.hostname</name>
+ <value>localhost</value>
+ </property>
+ <property>
+ <name>yarn.resourcemanager.bind-host</name>
+ <value>0.0.0.0</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.bind-host</name>
+ <value>0.0.0.0</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.resource.memory-mb</name>
+ <value>8192</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.resource.cpu-vcores</name>
+ <value>4</value>
+ </property>
+ <property>
+ <name>yarn.scheduler.minimum-allocation-mb</name>
+ <value>256</value>
+ </property>
+ <property>
+ <name>yarn.scheduler.maximum-allocation-mb</name>
+ <value>8192</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.pmem-check-enabled</name>
+ <value>false</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.vmem-check-enabled</name>
+ <value>false</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.env-whitelist</name>
+ <value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_HOME,PATH,LANG,TZ</value>
+ </property>
+ <property>
+ <name>yarn.nodemanager.delete.debug-delay-sec</name>
+ <value>600</value>
+ </property>
+</configuration>
diff --git a/docker/integration-tests/spark-yarn/start.sh b/docker/integration-tests/spark-yarn/start.sh
new file mode 100644
index 0000000000..4bc11a9edd
--- /dev/null
+++ b/docker/integration-tests/spark-yarn/start.sh
@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Starts HDFS (namenode + datanode) and YARN (resourcemanager + nodemanager) in one container.
+set -euo pipefail
+
+if [ ! -d /tmp/hadoop/dfs/name/current ]; then
+ hdfs namenode -format -force -nonInteractive
+fi
+hdfs --daemon start namenode
+hdfs --daemon start datanode
+yarn --daemon start resourcemanager
+yarn --daemon start nodemanager
+
+# Wait for HDFS, then create the folders Spark and the tests use
+until hdfs dfsadmin -safemode get 2>/dev/null | grep -q OFF; do sleep 2; done
+hdfs dfs -mkdir -p /user /tmp /data /output
+hdfs dfs -chmod -R 1777 /user /tmp /data /output
+echo "HDFS and YARN are up"
+
+tail -F "${HADOOP_LOG_DIR:-/opt/hadoop/logs}"/*.log
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc b/docs/hop-user-manual/modules/ROOT/nav.adoc
index 29e5a12eec..95ff2ed48d 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -71,6 +71,7 @@ under the License.
*** xref:pipeline/beam/developing-beam-transforms.adoc[Developing a Beam transform]
** xref:pipeline/spark/getting-started-with-native-spark.adoc[Getting started with the native Spark engine]
*** xref:pipeline/spark/local-standalone-cluster.adoc[Local standalone cluster]
+*** xref:pipeline/spark/yarn.adoc[YARN]
*** xref:pipeline/spark/paths-and-filesystems.adoc[Paths and file systems]
*** xref:pipeline/spark/cluster-mapping-walkthrough.adoc[Cluster mapping walk-through]
*** xref:pipeline/spark/lakehouse.adoc[Lakehouse tables (Delta / Iceberg)]
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc
index 325e0e7ff9..9b98824b65 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc
@@ -50,9 +50,9 @@ Use **Load configuration template** to fill the fields below from a common deplo
* *Local execution* — `local[*]`, Spark UI off (embedded Spark in Hop)
* *Local (capped cores)* — `local[4]`
* *Spark standalone server* — `spark://host:7077` (replace host), sample driver/executor memory. Also set *Fat jar file location* to a `native-provided` fat jar; the template leaves it empty. See xref:pipeline/spark/local-standalone-cluster.adoc[Local standalone cluster].
-* *spark-submit (cluster)* — master **empty** so `spark-submit --master` wins; use a `native-provided` fat jar
+* *spark-submit (cluster)* — master **empty** so `spark-submit --master` wins; use a `native-provided` fat jar. Outside `spark-submit` (Hop GUI, `hop-run`, Hop Server) an empty master runs on `local[*]` in that process, and the log warns about it.
* *Databricks* — submit-oriented placeholders (master empty); set workspace-specific values yourself
-* *YARN client* — master `yarn`, client deploy mode
+* *YARN client* — master `yarn`, client deploy mode. Also set *Hadoop configuration folder* (or `HADOOP_CONF_DIR`) and a `native-provided` fat jar. See xref:pipeline/spark/yarn.adoc[Run native Spark pipelines on YARN].
* *Lakehouse local* — same as local; Delta/Iceberg connectors live on the plugin classpath
Templates replace the current engine options after confirmation when the form already has custom values. The form fields update immediately after you apply a template. Edit values afterward as needed.
@@ -64,7 +64,7 @@ For named Iceberg/Delta catalogs used by Lake Table transforms in **TABLE** mode
|Option |Description |Default
|Spark master
-|Spark master URL: `local[*]`, `local[4]`, or `spark://host:7077`. Leave empty under `spark-submit` so the CLI master is used.
+|Spark master URL: `local[*]`, `local[4]`, `spark://host:7077` or `yarn`. Leave empty under `spark-submit` so the CLI master is used; outside `spark-submit` an empty master runs on `local[*]`.
|`local[*]`
|Application name
@@ -75,6 +75,10 @@ For named Iceberg/Delta catalogs used by Lake Table transforms in **TABLE** mode
|Optional Hop fat jar staged for executors when Hop starts the Spark session. Leave empty for pure `spark-submit` with a native-provided fat jar.
|
+|Hadoop configuration folder
+|Folder with the Hadoop client configuration (`core-site.xml`, `hdfs-site.xml`, `yarn-site.xml`) for sessions Hop starts itself. Its properties are passed to Spark as `spark.hadoop.*`; `spark.hadoop.*` lines in *Spark config* win. When empty and the master is `yarn`, Hop reads the folder in `HADOOP_CONF_DIR` or `YARN_CONF_DIR`. Not used under `spark-submit`. See xref:pipeline/spark/yarn.adoc[Run native Spark pipelines on YARN].
+|
+
|Spark config (key=value lines)
|Additional SparkConf entries, one `key=value` per line.
|
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
index 7e168a13ca..e8cf2a44ce 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
@@ -40,6 +40,9 @@ xref:pipeline/spark/paths-and-filesystems.adoc#spark-project-package[Spark proje
To run pipelines from Hop GUI or `hop-run` on a Spark standalone cluster on your own machine:
xref:pipeline/spark/local-standalone-cluster.adoc[Local standalone cluster].
+To run pipelines on a Hadoop YARN cluster from Hop GUI, `hop-run` or Hop Server:
+xref:pipeline/spark/yarn.adoc[Run native Spark pipelines on YARN].
+
Hands-on multi-node demo (Docker Compose + sample project `spark-demo` — Simple Mapping and Workflow Executor):
xref:pipeline/spark/cluster-mapping-walkthrough.adoc[Cluster demo walk-through].
@@ -108,7 +111,8 @@ If you need streaming, windowing, or Beam-only connectors, stay on the Beam engi
. For each transform it either:
** applies a *native handler* that rewrites the hop as Spark Dataset operations, or
** wraps the transform in `mapPartitions` so each Spark partition runs that transform in a tiny local Hop pipeline.
-. File outputs and other Spark *actions* materialise the graph (read → transform → write).
+. Spark File and Lake table outputs write while the graph is built. At the end the engine runs an action on every transform without a successor, so every branch runs, including branches that end in a wrapped output such as Text File Output or Table Output.
+. The output of a transform that several transforms read, such as the true and false targets of Filter Rows or the cases of Switch / Case, is cached: that transform runs once, not once per branch.
Each transform's reference page says which of the two it gets: a double check on the *Native Spark* tag means the engine has its own implementation (xref:pipeline/transforms/sort.adoc[Sort rows], xref:pipeline/transforms/mergejoin.adoc[Merge join], xref:pipeline/transforms/memgroupby.adoc[Memory group by], xref:pipeline/transforms/uniquerows.adoc[Unique rows], the Spark file and lake table transforms and xref:pipeline/transforms/spark-sql.adoc[Spark SQL]); a single check means the Hop tr [...]
. Transform metrics are collected via Spark accumulators and shown in Hop like any other engine.
@@ -284,7 +288,7 @@ Keep this export out of git if it is a point-in-time dump of secrets or large me
. Build the *native-provided* fat jar and export metadata (previous sections).
. Ensure the pipeline run configuration used on the cluster:
** Engine = *Native Spark pipeline engine* (not Beam Spark)
-** *Master* blank (so `spark-submit --master` wins) *or* set to `spark://host:7077`
+** *Master* blank (so `spark-submit --master` wins) *or* set to `spark://host:7077`. For YARN, submit with `--master yarn` from a machine where `HADOOP_CONF_DIR` points at the cluster's configuration; `spark-submit` reads it from there.
** *Fat jar* field empty (the submit jar already carries Hop)
. Submit:
@@ -303,8 +307,6 @@ export SPARK_HOME=/path/to/spark-4.1.x-bin-hadoop3
/path/to/cluster-env-config.json
----
-NOTE: Start `spark-submit` from a folder that does not contain a Hop `plugins` folder, not from the Hop installation folder. Otherwise Hop loads the Spark engine twice, from the fat jar and from that folder, and the run fails with `Run configuration '...' is not a native Spark pipeline engine configuration`. See xref:pipeline/spark/local-standalone-cluster.adoc#troubleshooting[Troubleshooting].
-
*Main class:* `org.apache.hop.spark.run.MainSpark`
*Positional arguments:*
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/local-standalone-cluster.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/local-standalone-cluster.adoc
index 353ee26fe6..f6774c009f 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/local-standalone-cluster.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/local-standalone-cluster.adoc
@@ -27,7 +27,7 @@ under the License.
This page sets up such a cluster on one machine and runs a pipeline on it from Hop GUI or `hop-run`. Hop is the Spark *driver*: you design and start the pipeline in Hop as usual, and the work runs on the workers.
-Do this after your first `local[*]` run (see xref:pipeline/spark/getting-started-with-native-spark.adoc[Getting started with the native Spark engine]). To submit with `spark-submit` inside Docker containers instead, see the xref:pipeline/spark/cluster-mapping-walkthrough.adoc[cluster demo walk-through].
+Do this after your first `local[*]` run (see xref:pipeline/spark/getting-started-with-native-spark.adoc[Getting started with the native Spark engine]). To submit with `spark-submit` inside Docker containers instead, see the xref:pipeline/spark/cluster-mapping-walkthrough.adoc[cluster demo walk-through]. To run on a Hadoop YARN cluster, see xref:pipeline/spark/yarn.adoc[Run native Spark pipelines on YARN].
== What you need
@@ -212,7 +212,7 @@ With `hop-run`, every run is a new Spark application. Even this small pipeline t
From Hop GUI, the first run starts the application and Hop GUI keeps it running, with all its cores, until you close Hop GUI. Later runs from Hop GUI reuse it. Close Hop GUI before you run on the cluster with `hop-run` or through Hop Server (step 7), or those runs wait for cores.
-The Spark application UI (jobs, stages and tasks, normally on port 4040) is not available: Hop turns it off for the Spark sessions it starts.
+The Spark application UI (jobs, stages and tasks) is off by default. Add `spark.ui.enabled=true` to *Spark config* to see it at http://localhost:4040 while the application runs.
The log shows the session connecting to the cluster:
@@ -254,7 +254,7 @@ Port 8181 because the Spark master UI already uses 8080 and the workers use 8081
** *Workflow engine type*: *Hop remote workflow engine*
** *Hop server*: `spark-hop-server`
** *Run Configuration*: `local`
-** *Export linked resources to server*: on. The workflow and pipeline are then sent as files. With it off, the samples project's Workflow Log fails every workflow you send to Hop Server (see <<troubleshooting>>).
+** *Export linked resources to server*: on. The workflow and pipeline are then sent as files. With it off, the samples project's Workflow Log fails every workflow you send to Hop Server (see <<troubleshooting>>). Export works here because Hop Server and the Spark workers share one machine: Hop Server runs the pipeline from a zip file in its temporary folder. With workers on other machines, turn export off and enable the project on Hop Server, as described for xref:pipeline/spark/yarn.ado [...]
. Create a workflow and save it as `spark/run-customers-per-state.hwf`: a *Start* action, then a *Pipeline* action with *Pipeline* `+${PROJECT_HOME}/spark/customers-per-state.hpl+` and *Run configuration* `spark-standalone`.
. Run the workflow with the `remote-server` run configuration, from Hop GUI or from the Hop installation folder:
+
@@ -312,7 +312,7 @@ Every executor copies the fat jar twice:
|`[PATH_NOT_FOUND] Path does not exist` from Spark file input
|The path must exist on the machine where each executor runs. Check the variables the path uses.
-|The log shows `Prepared native Spark pipeline engine` with the master you expect, then `Reusing active SparkSession` or `Reusing default SparkSession`, and the run uses a different master or an older application
+|The log shows `WARNING: run configuration '...' asks for Spark master '...', but this process already has a Spark session on '...'`
|An earlier native Spark run in the same Hop GUI, Hop Server or workflow left its Spark session running, and this run reuses it with that run's master, fat jar and Spark config. Restart Hop GUI or Hop Server after switching between native Spark run configurations.
|After a run from Hop GUI or through Hop Server, the application stays under *Running Applications* in the Spark master UI and keeps its cores
@@ -324,8 +324,8 @@ Every executor copies the fat jar twice:
|A workflow on Hop Server fails at the start with `NullPointerException: Cannot invoke "String.replaceAll(String, String)" because "vfsFilename" is null`
|The project has a *Workflow Log* that lists specific workflows (the samples project does), and the workflow was sent without its file. Turn on *Export linked resources to server* in the remote workflow run configuration, or disable the Workflow Log.
-|`spark-submit` fails with `Run configuration '...' is not a native Spark pipeline engine configuration (found org.apache.hop.spark.engines.SparkPipelineRunConfiguration)`
-|`spark-submit` was started from a folder that contains a Hop `plugins` folder, such as the Hop installation folder. Hop then loads the Spark engine twice: from the fat jar and from that folder. Start `spark-submit` from another folder.
+|The log shows `WARNING: run configuration '...' has no Spark master and this pipeline does not run under spark-submit`
+|An empty *Spark master* means `local[*]` inside the Hop process (Hop GUI, `hop-run` or Hop Server). Set the master, for example `spark://localhost:7077`. The *spark-submit (cluster)* template leaves it empty on purpose, for `spark-submit`.
|===
== Related pages
@@ -334,5 +334,6 @@ Every executor copies the fat jar twice:
* xref:pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc[Native Spark pipeline engine] (option reference)
* xref:pipeline/spark/paths-and-filesystems.adoc[Paths and file systems on native Spark]
* xref:pipeline/spark/cluster-mapping-walkthrough.adoc[Cluster demo walk-through] (Docker, `spark-submit`)
+* xref:pipeline/spark/yarn.adoc[Run native Spark pipelines on YARN]
* xref:hop-tools/hop-marketplace.adoc[Hop Marketplace]
* https://spark.apache.org/docs/4.1.3/spark-standalone.html[Spark standalone mode] (Apache Spark documentation)
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/yarn.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/yarn.adoc
new file mode 100644
index 0000000000..9d667a62e0
--- /dev/null
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/yarn.adoc
@@ -0,0 +1,182 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+////
+[[NativeSparkYarn]]
+:imagesdir: ../../../assets/images
+:description: Run native Spark pipelines on a Hadoop YARN cluster from Hop GUI, hop-run or Hop Server, with Hop as the Spark driver.
+:page-marketplace: hop-engines-spark
+
+:toc:
+
+= Run native Spark pipelines on YARN
+
+The native Spark engine can run a pipeline on a Hadoop YARN cluster while Hop itself is the Spark driver: you start the pipeline from Hop GUI, `hop-run` or Hop Server, and the executors run in YARN containers. Spark calls this *client deploy mode*.
+
+The alternative is `spark-submit --master yarn` on a machine with a Spark installation, with a fat jar and a project package; see xref:pipeline/spark/getting-started-with-native-spark.adoc[Getting started with the native Spark engine].
+
+Set up a xref:pipeline/spark/local-standalone-cluster.adoc[local standalone cluster] first if you have not run a pipeline on a cluster before: the fat jar, the run configuration and the troubleshooting there apply to YARN as well.
+
+== What you need
+
+* Hop 2.20 or later with the `hop-engines-spark` and `hop-engines-beam` plugins (see xref:pipeline/spark/local-standalone-cluster.adoc[Local standalone cluster], steps 1 and 3), and a `native-provided` fat jar.
+* The Hadoop client configuration of the cluster on the Hop machine: a folder with `core-site.xml`, `hdfs-site.xml` and `yarn-site.xml`. On a cluster edge node this is often `/etc/hadoop/conf`.
+* Network access from the Hop machine to the YARN ResourceManager and to HDFS (namenode and datanodes), and from the YARN nodes back to the Hop machine: the executors connect to the Spark driver, which is Hop.
+* Java 21 on the YARN nodes, for the Spark processes in the containers. If the nodes run an older Java by default, point the containers at a Java 21 installation (see below).
+* Data that the YARN nodes can read and write, for example on HDFS. The executors cannot read files on the Hop machine.
+
+Kerberos-secured clusters are not covered by this page.
+
+== Create the run configuration
+
+Create a *Pipeline run configuration* with *Engine type* *Native Spark pipeline engine*, and use *Load configuration template* -> *YARN client* as a start. Then set:
+
+[options="header",cols="1,2,3"]
+|===
+|Option |Value |Why
+
+|Spark master
+|`yarn`
+|The ResourceManager address comes from `yarn-site.xml`.
+
+|Hadoop configuration folder
+|The folder with the cluster's `*-site.xml` files, for example `/etc/hadoop/conf`
+|Leave it empty to use the `HADOOP_CONF_DIR` or `YARN_CONF_DIR` environment variable of the Hop process. Hop reads these files itself: unlike `spark-submit`, a Hop JVM does not put that folder on its classpath.
+
+|Fat jar file location
+|The absolute path of a `native-provided` fat jar
+|The executors load Hop from it.
+
+|Executor memory / Executor cores
+|For example `1g` / `1`
+|Must fit in a YARN container (`yarn.scheduler.maximum-allocation-mb` and `-vcores`).
+
+|Spark config (key=value lines)
+|`spark.ui.enabled=false` +
+`spark.submit.deployMode=client` +
+`spark.driver.host=<address of the Hop machine>` +
+`spark.executorEnv.JAVA_HOME=<Java 21 on the nodes>` +
+`spark.yarn.appMasterEnv.JAVA_HOME=<Java 21 on the nodes>`
+|`spark.driver.host` must be an address of the Hop machine that the YARN nodes can reach. The two `JAVA_HOME` lines are only needed when the nodes' default Java is older than 21.
+|===
+
+What Hop does with this on YARN:
+
+* It passes every property in the `*-site.xml` files to Spark as `spark.hadoop.<property>`. A `spark.hadoop.*` line in *Spark config* overrides the same property from the files.
+* It sets `spark.yarn.jars` to its own Spark runtime: the jars in `plugins/engines/spark/lib` plus the SLF4J and Jackson jars from `lib/core`, about 285 MB. Spark uploads them to the YARN staging folder on HDFS for every run, so the containers run exactly the Spark version Hop runs (4.1.3). If you set `spark.yarn.jars` or `spark.yarn.archive` in *Spark config* yourself, they must point at Spark 4.1.3 for Scala 2.13 and include those SLF4J and Jackson jars.
+
+Use HDFS paths (`hdfs:///data/...`) or other storage all nodes can reach in Spark File transforms; see xref:pipeline/spark/paths-and-filesystems.adoc[Paths and file systems].
+
+== Run it
+
+From Hop GUI or `hop-run`, run the pipeline with this run configuration. The log shows:
+
+[source]
+----
+Read <n> Hadoop configuration properties from file:///etc/hadoop/conf/core-site.xml, ...
+YARN containers get the engine's Spark runtime: <n> jars in spark.yarn.jars
+Created SparkSession version=4.1.3 master=yarn (owned)
+...
+Spark pipeline finished, result row count=0
+----
+
+The application appears in the YARN ResourceManager UI as `Apache Hop` (the *Application name* of the run configuration). Uploading the jars makes the start of every run take a while; a small pipeline takes about a minute.
+
+Like on a standalone cluster, Hop GUI and Hop Server keep their Spark application, and the YARN resources it holds, until they stop; `hop-run` ends it when the pipeline is done.
+
+== Run it through Hop Server
+
+Run a workflow on Hop Server whose *Pipeline* action uses the YARN run configuration, as described in xref:pipeline/spark/local-standalone-cluster.adoc#hop-server[Run it through Hop Server], with two differences:
+
+* Set the Hadoop configuration folder in the run configuration, or start Hop Server with `HADOOP_CONF_DIR` set.
+* Turn *Export linked resources to server* *off* in the remote workflow run configuration, and enable the project on Hop Server (`hop-server.sh -j <project>`). With export on, Hop Server runs the pipeline from a zip file in its own temporary folder, and the executors on the YARN nodes cannot open it.
+
+With export off, the samples project's Workflow Log fails every workflow you send to Hop Server; disable that Workflow Log or use your own project (see the xref:pipeline/spark/local-standalone-cluster.adoc#troubleshooting[troubleshooting table]).
+
+[[yarn-docker]]
+== Try it with a single-node YARN cluster in Docker
+
+The Hop source tree has a single-node HDFS and YARN cluster for Docker on Linux: `docker/integration-tests/integration-tests-spark-native-yarn.yaml`. Its Hadoop configuration in `docker/integration-tests/spark-yarn/conf` serves both the container and Hop on the Docker host.
+
+. Start it from the root of the Hop source tree:
++
+[source,bash]
+----
+docker compose -f docker/integration-tests/integration-tests-spark-native-yarn.yaml up -d --build
+----
++
+The YARN UI is at http://localhost:8088 and HDFS at http://localhost:9870.
+. Copy the sample data to HDFS:
++
+[source,bash]
+----
+C=$(docker compose -f docker/integration-tests/integration-tests-spark-native-yarn.yaml ps -q hadoop-yarn)
+docker cp /path/to/hop/config/projects/samples/beam/input/customers-noheader-1k.txt $C:/tmp/
+docker exec $C hdfs dfs -put -f /tmp/customers-noheader-1k.txt /data/
+----
+. Create the run configuration with *Hadoop configuration folder* `<hop source>/docker/integration-tests/spark-yarn/conf`, your fat jar, and these *Spark config* lines:
++
+[source]
+----
+spark.ui.enabled=false
+spark.submit.deployMode=client
+spark.driver.host=172.31.250.1
+spark.executorEnv.JAVA_HOME=/opt/java21
+spark.yarn.appMasterEnv.JAVA_HOME=/opt/java21
+----
++
+`172.31.250.1` is the Docker host on the cluster's network, and `/opt/java21` is the Java 21 runtime in the image.
+. Use the pipeline from the xref:pipeline/spark/local-standalone-cluster.adoc[local cluster guide] with the paths `hdfs:///data/customers-noheader-1k.txt` and `hdfs:///output/customers-per-state`, and run it.
+. Check the result:
++
+[source,bash]
+----
+docker exec $C bash -c 'hdfs dfs -cat /output/customers-per-state/part-*.csv'
+----
+
+Stop the cluster with `docker compose -f docker/integration-tests/integration-tests-spark-native-yarn.yaml down`.
+
+== Troubleshooting
+
+[options="header",cols="2,3"]
+|===
+|Symptom |Cause and fix
+
+|The log shows `WARNING: Spark master is yarn, but no Hadoop configuration folder is set` and Spark keeps retrying `Connecting to ResourceManager at /0.0.0.0:8032`
+|Hop found no Hadoop configuration. Set *Hadoop configuration folder*, or set `HADOOP_CONF_DIR` for the Hop process (Hop GUI, `hop-run` or Hop Server).
+
+|`File /user/<you>/.sparkStaging/... could only be written to 0 of the 1 minReplication nodes`
+|The Hop machine cannot reach the HDFS datanodes to upload the Spark jars. Check the datanode addresses the namenode hands out and the network between the Hop machine and the datanodes.
+
+|`The YARN application has already ended! It might have been killed or the Application Master may have failed to start`
+|Look at the logs of the application's first container in the YARN UI. `UnsupportedClassVersionError: org/apache/spark/deploy/yarn/ExecutorLauncher has been compiled by a more recent version of the Java Runtime` means the nodes run an older Java: set `spark.executorEnv.JAVA_HOME` and `spark.yarn.appMasterEnv.JAVA_HOME` to a Java 21 installation on the nodes.
+
+|`NoClassDefFoundError: org/slf4j/LoggerFactory` in the first container
+|`spark.yarn.jars` or `spark.yarn.archive` in *Spark config* leave out SLF4J or Jackson. Remove them to use Hop's default, or add those jars.
+
+|`Could not replicate "file:///tmp/export_...zip" as it does not exist` from a transform on Hop Server
+|The workflow was sent with *Export linked resources to server* on. Turn it off and enable the project on Hop Server.
+
+|`Could not parse Master URL: 'yarn'`
+|The Spark engine plugin is older than 2.20 and has no YARN support. Update the `hop-engines-spark` plugin.
+|===
+
+== Related pages
+
+* xref:pipeline/spark/local-standalone-cluster.adoc[Run pipelines on a local Spark standalone cluster]
+* xref:pipeline/spark/getting-started-with-native-spark.adoc[Getting started with the native Spark engine]
+* xref:pipeline/pipeline-run-configurations/native-spark-pipeline-engine.adoc[Native Spark pipeline engine] (option reference)
+* xref:pipeline/spark/paths-and-filesystems.adoc[Paths and file systems on native Spark]
+* https://spark.apache.org/docs/4.1.3/running-on-yarn.html[Running Spark on YARN] (Apache Spark documentation)
diff --git a/integration-tests/spark-native/0010-filter-text-outputs-validation.hpl b/integration-tests/spark-native/0010-filter-text-outputs-validation.hpl
new file mode 100644
index 0000000000..a3cb2e7569
--- /dev/null
+++ b/integration-tests/spark-native/0010-filter-text-outputs-validation.hpl
@@ -0,0 +1,328 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<pipeline>
+ <info>
+ <name>0010-filter-text-outputs-validation</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Validate dual filter-target CSV outputs</description>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters/>
+ <capture_transform_performance>N</capture_transform_performance>
+ <created_user>-</created_user>
+ <created_date>2026/07/15 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/15 00:00:00.000</modified_date>
+ </info>
+ <notepads/>
+ <order>
+ <hop><from>Read true CSV</from><to>Verify true</to><enabled>Y</enabled></hop>
+ <hop><from>Read false CSV</from><to>Verify false</to><enabled>Y</enabled></hop>
+ </order>
+ <transform>
+ <name>Read true CSV</name>
+ <type>TextFileInput2</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <accept_filenames>N</accept_filenames>
+ <passing_through_fields>N</passing_through_fields>
+ <accept_field/>
+ <accept_transform_name/>
+ <separator>,</separator>
+ <enclosure>"</enclosure>
+ <enclosure_breaks>N</enclosure_breaks>
+ <escapechar/>
+ <header>Y</header>
+ <nr_headerlines>1</nr_headerlines>
+ <footer>N</footer>
+ <nr_footerlines>1</nr_footerlines>
+ <line_wrapped>N</line_wrapped>
+ <nr_wraps>1</nr_wraps>
+ <layout_paged>N</layout_paged>
+ <nr_lines_per_page>80</nr_lines_per_page>
+ <nr_lines_doc_header>0</nr_lines_doc_header>
+ <noempty>Y</noempty>
+ <include>N</include>
+ <include_field/>
+ <rownum>N</rownum>
+ <rownumByFile>N</rownumByFile>
+ <rownum_field/>
+ <format>Unix</format>
+ <encoding/>
+ <length>Characters</length>
+ <add_to_result_filenames>Y</add_to_result_filenames>
+ <file>
+ <name>${PROJECT_HOME}/output/0010-filter-true/</name>
+ <filemask>.*\.csv</filemask>
+ <exclude_filemask/>
+ <file_required>Y</file_required>
+ <include_subfolders>N</include_subfolders>
+ <type>CSV</type>
+ <compression>None</compression>
+ </file>
+ <filters/>
+ <fields>
+ <field>
+ <name>id</name>
+ <type>Integer</type>
+ <format>#</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ <field>
+ <name>stateCode</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ <field>
+ <name>state</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ </fields>
+ <limit>0</limit>
+ <error_ignored>N</error_ignored>
+ <skip_bad_files>N</skip_bad_files>
+ <file_error_field/>
+ <file_error_message_field/>
+ <error_line_skipped>N</error_line_skipped>
+ <error_count_field/>
+ <error_fields_field/>
+ <error_text_field/>
+ <bad_line_files_destination_directory/>
+ <bad_line_files_extension>warning</bad_line_files_extension>
+ <error_line_files_destination_directory/>
+ <error_line_files_extension>error</error_line_files_extension>
+ <line_number_files_destination_directory/>
+ <line_number_files_extension>line</line_number_files_extension>
+ <date_format_lenient>Y</date_format_lenient>
+ <date_format_locale/>
+ <shortFileFieldName/>
+ <pathFieldName/>
+ <hiddenFieldName/>
+ <lastModificationTimeFieldName/>
+ <uriNameFieldName/>
+ <rootUriNameFieldName/>
+ <extensionFieldName/>
+ <sizeFieldName/>
+ <attributes/>
+
+ <GUI>
+ <xloc>128</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+
+ <transform>
+ <name>Verify true</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+
+ <transform>
+ <name>Read false CSV</name>
+ <type>TextFileInput2</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <accept_filenames>N</accept_filenames>
+ <passing_through_fields>N</passing_through_fields>
+ <accept_field/>
+ <accept_transform_name/>
+ <separator>,</separator>
+ <enclosure>"</enclosure>
+ <enclosure_breaks>N</enclosure_breaks>
+ <escapechar/>
+ <header>Y</header>
+ <nr_headerlines>1</nr_headerlines>
+ <footer>N</footer>
+ <nr_footerlines>1</nr_footerlines>
+ <line_wrapped>N</line_wrapped>
+ <nr_wraps>1</nr_wraps>
+ <layout_paged>N</layout_paged>
+ <nr_lines_per_page>80</nr_lines_per_page>
+ <nr_lines_doc_header>0</nr_lines_doc_header>
+ <noempty>Y</noempty>
+ <include>N</include>
+ <include_field/>
+ <rownum>N</rownum>
+ <rownumByFile>N</rownumByFile>
+ <rownum_field/>
+ <format>Unix</format>
+ <encoding/>
+ <length>Characters</length>
+ <add_to_result_filenames>Y</add_to_result_filenames>
+ <file>
+ <name>${PROJECT_HOME}/output/0010-filter-false/</name>
+ <filemask>.*\.csv</filemask>
+ <exclude_filemask/>
+ <file_required>Y</file_required>
+ <include_subfolders>N</include_subfolders>
+ <type>CSV</type>
+ <compression>None</compression>
+ </file>
+ <filters/>
+ <fields>
+ <field>
+ <name>id</name>
+ <type>Integer</type>
+ <format>#</format>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ <field>
+ <name>stateCode</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ <field>
+ <name>state</name>
+ <type>String</type>
+ <format/>
+ <currency/>
+ <decimal/>
+ <group/>
+ <nullif/>
+ <ifnull/>
+ <position>-1</position>
+ <length>-1</length>
+ <precision>-1</precision>
+ <trim_type>both</trim_type>
+ <repeat>N</repeat>
+ </field>
+ </fields>
+ <limit>0</limit>
+ <error_ignored>N</error_ignored>
+ <skip_bad_files>N</skip_bad_files>
+ <file_error_field/>
+ <file_error_message_field/>
+ <error_line_skipped>N</error_line_skipped>
+ <error_count_field/>
+ <error_fields_field/>
+ <error_text_field/>
+ <bad_line_files_destination_directory/>
+ <bad_line_files_extension>warning</bad_line_files_extension>
+ <error_line_files_destination_directory/>
+ <error_line_files_extension>error</error_line_files_extension>
+ <line_number_files_destination_directory/>
+ <line_number_files_extension>line</line_number_files_extension>
+ <date_format_lenient>Y</date_format_lenient>
+ <date_format_locale/>
+ <shortFileFieldName/>
+ <pathFieldName/>
+ <hiddenFieldName/>
+ <lastModificationTimeFieldName/>
+ <uriNameFieldName/>
+ <rootUriNameFieldName/>
+ <extensionFieldName/>
+ <sizeFieldName/>
+ <attributes/>
+
+ <GUI>
+ <xloc>128</xloc>
+ <yloc>224</yloc>
+ </GUI>
+ </transform>
+
+ <transform>
+ <name>Verify false</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>320</xloc>
+ <yloc>224</yloc>
+ </GUI>
+ </transform>
+
+ <attributes/>
+ <transform_error_handling/>
+</pipeline>
diff --git a/integration-tests/spark-native/0010-filter-text-outputs.hpl b/integration-tests/spark-native/0010-filter-text-outputs.hpl
new file mode 100644
index 0000000000..e3b6c36f9a
--- /dev/null
+++ b/integration-tests/spark-native/0010-filter-text-outputs.hpl
@@ -0,0 +1,207 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<pipeline>
+ <info>
+ <name>0010-filter-text-outputs</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Filter Rows targets that end in Text File Output: both branches must be written</description>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters/>
+ <capture_transform_performance>N</capture_transform_performance>
+ <created_user>-</created_user>
+ <created_date>2026/07/15 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/15 00:00:00.000</modified_date>
+ </info>
+ <notepads/>
+ <order>
+ <hop><from>Spark File Input</from><to>State starts with NO</to><enabled>Y</enabled></hop>
+ <hop><from>State starts with NO</from><to>True: select</to><enabled>Y</enabled></hop>
+ <hop><from>State starts with NO</from><to>False: select</to><enabled>Y</enabled></hop>
+ <hop><from>True: select</from><to>output/true</to><enabled>Y</enabled></hop>
+ <hop><from>False: select</from><to>output/false</to><enabled>Y</enabled></hop>
+ </order>
+ <transform>
+ <type>SparkFileInput</type>
+ <name>Spark File Input</name>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning><method>none</method><schema_name/></partitioning>
+ <file_path>${PROJECT_HOME}/files/customers-1k.txt</file_path>
+ <file_format>csv</file_format>
+ <header>Y</header>
+ <separator>;</separator>
+ <quote>"</quote>
+ <infer_schema>N</infer_schema>
+ <multi_line>N</multi_line>
+ <extra_options/>
+ <fields>
+ <field><name>id</name><type>Integer</type><length>7</length><precision>-1</precision></field>
+ <field><name>name</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>firstname</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>zip</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>city</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>birthdate</name><type>String</type><length>20</length><precision>-1</precision></field>
+ <field><name>street</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>housenr</name><type>String</type><length>50</length><precision>-1</precision></field>
+ <field><name>stateCode</name><type>String</type><length>-1</length><precision>-1</precision></field>
+ <field><name>state</name><type>String</type><length>50</length><precision>-1</precision></field>
+ </fields>
+ <attributes/>
+ <GUI><xloc>96</xloc><yloc>160</yloc></GUI>
+ </transform>
+ <transform>
+ <type>FilterRows</type>
+ <name>State starts with NO</name>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning><method>none</method><schema_name/></partitioning>
+ <compare>
+ <condition>
+ <negated>N</negated>
+ <operator>-</operator>
+ <leftvalue>state</leftvalue>
+ <function>STARTS WITH</function>
+ <value>
+ <name>constant</name>
+ <type>String</type>
+ <text>NO</text>
+ <length>-1</length>
+ <precision>-1</precision>
+ <isnull>N</isnull>
+ <mask/>
+ </value>
+ <conditions/>
+ </condition>
+ </compare>
+ <send_true_to>True: select</send_true_to>
+ <send_false_to>False: select</send_false_to>
+ <attributes/>
+ <GUI><xloc>288</xloc><yloc>160</yloc></GUI>
+ </transform>
+ <transform>
+ <type>SelectValues</type>
+ <name>True: select</name>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning><method>none</method><schema_name/></partitioning>
+ <fields>
+ <field><name>id</name><rename/><length>-2</length><precision>-2</precision></field>
+ <field><name>stateCode</name><rename/><length>-2</length><precision>-2</precision></field>
+ <field><name>state</name><rename/><length>-2</length><precision>-2</precision></field>
+ <select_unspecified>N</select_unspecified>
+ </fields>
+ <attributes/>
+ <GUI><xloc>480</xloc><yloc>96</yloc></GUI>
+ </transform>
+ <transform>
+ <type>SelectValues</type>
+ <name>False: select</name>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning><method>none</method><schema_name/></partitioning>
+ <fields>
+ <field><name>id</name><rename/><length>-2</length><precision>-2</precision></field>
+ <field><name>stateCode</name><rename/><length>-2</length><precision>-2</precision></field>
+ <field><name>state</name><rename/><length>-2</length><precision>-2</precision></field>
+ <select_unspecified>N</select_unspecified>
+ </fields>
+ <attributes/>
+ <GUI><xloc>480</xloc><yloc>224</yloc></GUI>
+ </transform>
+ <transform>
+ <name>output/true</name>
+ <type>TextFileOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <separator>,</separator>
+ <enclosure>"</enclosure>
+ <header>Y</header>
+ <footer>N</footer>
+ <format>UNIX</format>
+ <compression>None</compression>
+ <encoding>UTF-8</encoding>
+ <fileNameInField>N</fileNameInField>
+ <create_parent_folder>Y</create_parent_folder>
+ <file>
+ <name>${PROJECT_HOME}/output/0010-filter-true/true-${Internal.Transform.ID}</name>
+ <servlet_output>N</servlet_output>
+ <do_not_open_new_file_init>N</do_not_open_new_file_init>
+ <extention>csv</extention>
+ <append>N</append>
+ <split>N</split>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <SpecifyFormat>N</SpecifyFormat>
+ <add_to_result_filenames>N</add_to_result_filenames>
+ <pad>N</pad>
+ <fast_dump>N</fast_dump>
+ </file>
+ <fields/>
+ <attributes/>
+ <GUI><xloc>672</xloc><yloc>96</yloc></GUI>
+ </transform>
+ <transform>
+ <name>output/false</name>
+ <type>TextFileOutput</type>
+ <description/>
+ <distribute>Y</distribute>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <separator>,</separator>
+ <enclosure>"</enclosure>
+ <header>Y</header>
+ <footer>N</footer>
+ <format>UNIX</format>
+ <compression>None</compression>
+ <encoding>UTF-8</encoding>
+ <fileNameInField>N</fileNameInField>
+ <create_parent_folder>Y</create_parent_folder>
+ <file>
+ <name>${PROJECT_HOME}/output/0010-filter-false/false-${Internal.Transform.ID}</name>
+ <servlet_output>N</servlet_output>
+ <do_not_open_new_file_init>N</do_not_open_new_file_init>
+ <extention>csv</extention>
+ <append>N</append>
+ <split>N</split>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <SpecifyFormat>N</SpecifyFormat>
+ <add_to_result_filenames>N</add_to_result_filenames>
+ <pad>N</pad>
+ <fast_dump>N</fast_dump>
+ </file>
+ <fields/>
+ <attributes/>
+ <GUI><xloc>672</xloc><yloc>224</yloc></GUI>
+ </transform>
+ <attributes/>
+ <transform_error_handling/>
+</pipeline>
diff --git a/integration-tests/spark-native/main-0010-filter-text-outputs.hwf b/integration-tests/spark-native/main-0010-filter-text-outputs.hwf
new file mode 100644
index 0000000000..d833221940
--- /dev/null
+++ b/integration-tests/spark-native/main-0010-filter-text-outputs.hwf
@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<workflow>
+ <name>main-0010-filter-text-outputs</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Native Spark Filter Rows dual targets then golden unit test on Local</description>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/07/15 00:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/15 00:00:00.000</modified_date>
+ <parameters/>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <intervalSeconds>0</intervalSeconds>
+ <intervalMinutes>60</intervalMinutes>
+ <hour>12</hour>
+ <minutes>0</minutes>
+ <weekDay>1</weekDay>
+ <DayOfMonth>1</DayOfMonth>
+ <parallel>N</parallel>
+ <xloc>64</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>delete output</name>
+ <description/>
+ <type>DELETE_FILES</type>
+ <attributes/>
+ <arg_from_previous>N</arg_from_previous>
+ <include_subfolders>N</include_subfolders>
+ <fields>
+ <field>
+ <name>${PROJECT_HOME}/output/0010-filter-true/</name>
+ <filemask>.*</filemask>
+ </field>
+ <field>
+ <name>${PROJECT_HOME}/output/0010-filter-false/</name>
+ <filemask>.*</filemask>
+ </field>
+ </fields>
+ <parallel>N</parallel>
+ <xloc>224</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>0010-filter-text-outputs.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <filename>${PROJECT_HOME}/0010-filter-text-outputs.hpl</filename>
+ <params_from_previous>N</params_from_previous>
+ <exec_per_row>N</exec_per_row>
+ <clear_rows>N</clear_rows>
+ <clear_files>N</clear_files>
+ <set_logfile>N</set_logfile>
+ <logfile/>
+ <logext/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <loglevel>Basic</loglevel>
+ <set_append_logfile>N</set_append_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <create_parent_folder>Y</create_parent_folder>
+ <run_configuration>spark-local</run_configuration>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <parallel>N</parallel>
+ <xloc>416</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ <action>
+ <name>Run Pipeline Unit Tests</name>
+ <description/>
+ <type>RunPipelineTests</type>
+ <attributes/>
+ <test_names>
+ <test_name>
+ <name>0010-filter-text-outputs-validation UNIT</name>
+ </test_name>
+ </test_names>
+ <parallel>N</parallel>
+ <xloc>640</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ <hop>
+ <from>Start</from>
+ <to>delete output</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>Y</unconditional>
+ </hop>
+ <hop>
+ <from>delete output</from>
+ <to>0010-filter-text-outputs.hpl</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0010-filter-text-outputs.hpl</from>
+ <to>Run Pipeline Unit Tests</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ </hops>
+ <notepads/>
+ <attributes/>
+</workflow>
diff --git a/integration-tests/spark-native/metadata/unit-test/0010-filter-text-outputs-validation UNIT.json b/integration-tests/spark-native/metadata/unit-test/0010-filter-text-outputs-validation UNIT.json
new file mode 100644
index 0000000000..9b22147ca4
--- /dev/null
+++ b/integration-tests/spark-native/metadata/unit-test/0010-filter-text-outputs-validation UNIT.json
@@ -0,0 +1,57 @@
+{
+ "variableValues": [],
+ "database_replacements": [],
+ "autoOpening": true,
+ "basePath": "",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "id",
+ "data_set_field": "id"
+ },
+ {
+ "transform_field": "stateCode",
+ "data_set_field": "stateCode"
+ },
+ {
+ "transform_field": "state",
+ "data_set_field": "state"
+ }
+ ],
+ "field_order": [
+ "id"
+ ],
+ "data_set_name": "0005-filter-true-golden",
+ "transform_name": "Verify true"
+ },
+ {
+ "field_mappings": [
+ {
+ "transform_field": "id",
+ "data_set_field": "id"
+ },
+ {
+ "transform_field": "stateCode",
+ "data_set_field": "stateCode"
+ },
+ {
+ "transform_field": "state",
+ "data_set_field": "state"
+ }
+ ],
+ "field_order": [
+ "id"
+ ],
+ "data_set_name": "0005-filter-false-golden",
+ "transform_name": "Verify false"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0010-filter-text-outputs-validation UNIT",
+ "description": "Both Text File Output branches of a Filter Rows on native Spark are written",
+ "persist_filename": "",
+ "trans_test_tweaks": [],
+ "pipeline_filename": "./0010-filter-text-outputs-validation.hpl",
+ "test_type": "UNIT_TEST"
+}
\ No newline at end of file
diff --git a/plugins/engines/spark/pom.xml b/plugins/engines/spark/pom.xml
index 74675f27c0..03c2bb1c0f 100644
--- a/plugins/engines/spark/pom.xml
+++ b/plugins/engines/spark/pom.xml
@@ -144,6 +144,31 @@
</exclusion>
</exclusions>
</dependency>
+ <!-- YARN client for runs that Hop starts itself (Hop GUI, hop-run, Hop Server) with
+ master "yarn". spark-submit on a cluster brings its own. -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-yarn_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<!-- Maintained fork of org.lz4:lz4-java (same net.jpountz packages), shipped in lib/core. -->
<groupId>at.yawk.lz4</groupId>
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopMapPartitionsFn.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopMapPartitionsFn.java
index 262cbaf111..e48ab41d95 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopMapPartitionsFn.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopMapPartitionsFn.java
@@ -26,6 +26,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
import org.apache.hop.core.exception.HopException;
@@ -489,7 +490,8 @@ public class HopMapPartitionsFn implements MapPartitionsFunction<Row, Row>, Seri
&& !acceptFilenamesFromMain
&& !multiTarget
&& infoTransforms.isEmpty()
- && mainTransform instanceof BaseTransform;
+ && mainTransform instanceof BaseTransform
+ && !needsBatchComplete(transformPluginId);
if (direct) {
// output captured by SparkDirectRowHandler.putRow
} else if (!multiTarget) {
@@ -1218,4 +1220,15 @@ public class HopMapPartitionsFn implements MapPartitionsFunction<Row, Row>, Seri
return false;
}
}
+
+ /**
+ * Transforms that drive a child pipeline from {@code batchComplete()}. The direct row path calls
+ * {@code processRow()} per row and never {@code batchComplete()}, so these hang there and need
+ * the single-threaded executor path.
+ */
+ private static final Set<String> BATCH_COMPLETE_PLUGIN_IDS = Set.of("SimpleMapping");
+
+ static boolean needsBatchComplete(String transformPluginId) {
+ return BATCH_COMPLETE_PLUGIN_IDS.contains(transformPluginId);
+ }
}
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopSparkUtil.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopSparkUtil.java
index c4d425a6f0..a66cd32a85 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopSparkUtil.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopSparkUtil.java
@@ -49,6 +49,14 @@ public final class HopSparkUtil {
return transformName + " - TARGET - " + targetTransformName;
}
+ /**
+ * Dataset-map key for the Dataset that all target streams of a transform are filtered from. The
+ * converter persists it so the transform runs once, not once per branch.
+ */
+ public static String createSharedOutputId(String transformName) {
+ return transformName + " - SHARED OUTPUT";
+ }
+
public static void loadTransformMetadataFromXml(
String transformName,
ITransformMeta iTransformMeta,
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/ISparkPipelineEngineRunConfiguration.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/ISparkPipelineEngineRunConfiguration.java
index 19c9a241cb..48be7775e7 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/ISparkPipelineEngineRunConfiguration.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/ISparkPipelineEngineRunConfiguration.java
@@ -27,6 +27,13 @@ public interface ISparkPipelineEngineRunConfiguration extends IPipelineEngineRun
String getFatJar();
+ /**
+ * Optional folder with Hadoop client configuration ({@code core-site.xml}, {@code yarn-site.xml},
+ * ...) for sessions Hop starts itself. Empty: {@code HADOOP_CONF_DIR} / {@code YARN_CONF_DIR} for
+ * YARN masters.
+ */
+ String getHadoopConfDir();
+
String getSparkConfigs();
String getDriverMemory();
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
index d450fd7863..86bb346886 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
@@ -94,6 +94,7 @@ import org.apache.hop.spark.pipeline.HopPipelineMetaToSparkConverter;
import org.apache.hop.spark.pkg.SparkProjectPackage;
import org.apache.hop.spark.table.LakeSessionPlan;
import org.apache.hop.spark.util.SparkConst;
+import org.apache.hop.spark.util.SparkHadoopConfiguration;
import org.apache.hop.workflow.WorkflowMeta;
import org.apache.hop.workflow.engine.IWorkflowEngine;
import org.apache.spark.SparkConf;
@@ -340,8 +341,17 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
try {
Thread.currentThread().setContextClassLoader(pluginCl);
resultDataset = converter.createDataset(sparkSession);
- // Materialize: count is a safe action when no dedicated sink handler wrote data
- long rows = resultDataset.count();
+ // Generic transforms (outputs included) are lazy: run an action on every leaf,
+ // or only the branch of the last one runs. Spark file and lake table outputs
+ // already wrote while the graph was built; their leaves are empty markers.
+ long rows = 0;
+ for (Map.Entry<String, Dataset<Row>> leaf :
+ converter.getLeafDatasets().entrySet()) {
+ long leafRows = leaf.getValue().count();
+ logChannel.logDetailed(
+ "Transform '" + leaf.getKey() + "' finished, row count=" + leafRows);
+ rows += leafRows;
+ }
logChannel.logBasic("Spark pipeline finished, result row count=" + rows);
setErrors(0);
} catch (Throwable e) {
@@ -350,6 +360,7 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
setStopped(true);
} finally {
Thread.currentThread().setContextClassLoader(prev);
+ converter.unpersistAll();
setRunning(false);
setFinished(true);
executionEndDate = new Date();
@@ -1039,6 +1050,7 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
"Reusing active SparkSession (version="
+ session.version()
+ ", spark-submit or existing driver context)");
+ warnIfReusedMasterDiffers(session);
if (!lakePlan.isEmpty()) {
lakePlan.verifyActiveSession(session, logChannel, this);
}
@@ -1049,6 +1061,7 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
sessionOwnedByThisEngine = false;
SparkSession session = def.get();
logChannel.logBasic("Reusing default SparkSession (version=" + session.version() + ")");
+ warnIfReusedMasterDiffers(session);
if (!lakePlan.isEmpty()) {
lakePlan.verifyActiveSession(session, logChannel, this);
}
@@ -1070,7 +1083,19 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
// ignore — fall back to local
}
}
- master = StringUtils.isNotEmpty(confMaster) ? confMaster : "local[*]";
+ if (StringUtils.isNotEmpty(confMaster)) {
+ master = confMaster;
+ } else {
+ master = "local[*]";
+ if (!underSubmit) {
+ logChannel.logBasic(
+ "WARNING: run configuration '"
+ + pipelineRunConfiguration.getName()
+ + "' has no Spark master and this pipeline does not run under spark-submit: it"
+ + " runs on local[*] inside this process, not on a cluster. Set the Spark master"
+ + " (for example spark://host:7077 or yarn) to run on a cluster.");
+ }
+ }
}
String appName = resolve(sparkEngineRunConfiguration.getSparkAppName());
if (StringUtils.isEmpty(appName)) {
@@ -1103,6 +1128,14 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
}
}
+ // Hadoop client configuration and YARN defaults. Applied before the run configuration's Spark
+ // config so explicit spark.hadoop.* / spark.yarn.* lines there still win. spark-submit already
+ // puts HADOOP_CONF_DIR on the classpath and provides Spark on YARN.
+ if (!underSubmit) {
+ applyHadoopConfiguration(builder, master);
+ }
+
+ Set<String> configuredKeys = new HashSet<>();
String extra = sparkEngineRunConfiguration.getSparkConfigs();
if (StringUtils.isNotEmpty(extra)) {
for (String line : extra.split("\\r?\\n")) {
@@ -1112,8 +1145,9 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
}
int eq = trimmed.indexOf('=');
if (eq > 0) {
- builder.config(
- resolve(trimmed.substring(0, eq).trim()), resolve(trimmed.substring(eq + 1).trim()));
+ String key = resolve(trimmed.substring(0, eq).trim());
+ builder.config(key, resolve(trimmed.substring(eq + 1).trim()));
+ configuredKeys.add(key);
}
}
}
@@ -1126,10 +1160,14 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
}
// Defaults for local/dev only when not already set by submit or run config extras
- if (!underSubmit && System.getProperty("spark.ui.enabled") == null) {
+ if (!underSubmit
+ && System.getProperty("spark.ui.enabled") == null
+ && !configuredKeys.contains("spark.ui.enabled")) {
builder.config("spark.ui.enabled", "false");
}
- if (!underSubmit && System.getProperty("spark.sql.shuffle.partitions") == null) {
+ if (!underSubmit
+ && System.getProperty("spark.sql.shuffle.partitions") == null
+ && !configuredKeys.contains("spark.sql.shuffle.partitions")) {
builder.config("spark.sql.shuffle.partitions", "4");
}
@@ -1161,6 +1199,60 @@ public class SparkPipelineEngine extends Variables implements IPipelineEngine<Pi
return StringUtils.isNotEmpty(master) && !master.startsWith("local");
}
+ /**
+ * A session that an earlier run left in this process keeps the master, fat jar and Spark config
+ * of that run. Say so when this run configuration asks for another master.
+ */
+ private void warnIfReusedMasterDiffers(SparkSession session) {
+ String requested = resolve(sparkEngineRunConfiguration.getSparkMaster());
+ String actual = session.sparkContext().master();
+ if (StringUtils.isNotEmpty(requested) && !requested.trim().equals(actual)) {
+ logChannel.logBasic(
+ "WARNING: run configuration '"
+ + pipelineRunConfiguration.getName()
+ + "' asks for Spark master '"
+ + requested.trim()
+ + "', but this process already has a Spark session on '"
+ + actual
+ + "' from an earlier run. This pipeline runs on '"
+ + actual
+ + "' with that run's settings. Restart Hop GUI or Hop Server to switch.");
+ }
+ }
+
+ /**
+ * Passes the Hadoop client configuration to Spark as {@code spark.hadoop.*} and, on YARN, ships
+ * the engine's own Spark jars to the containers so they run the Spark version Hop runs.
+ */
+ private void applyHadoopConfiguration(SparkSession.Builder builder, String master)
+ throws HopException {
+ boolean yarn = SparkHadoopConfiguration.isYarnMaster(master);
+ List<String> folders =
+ SparkHadoopConfiguration.resolveFolders(
+ resolve(sparkEngineRunConfiguration.getHadoopConfDir()), yarn, System::getenv);
+ if (!folders.isEmpty()) {
+ Map<String, String> properties = SparkHadoopConfiguration.load(folders, logChannel);
+ for (Map.Entry<String, String> property : properties.entrySet()) {
+ builder.config(property.getKey(), property.getValue());
+ }
+ } else if (yarn) {
+ logChannel.logBasic(
+ "WARNING: Spark master is yarn, but no Hadoop configuration folder is set and"
+ + " HADOOP_CONF_DIR / YARN_CONF_DIR are not set. The YARN client uses its defaults"
+ + " (ResourceManager at 0.0.0.0:8032).");
+ }
+ if (yarn) {
+ String yarnJars = SparkHadoopConfiguration.defaultYarnJars(SparkPipelineEngine.class);
+ if (yarnJars != null) {
+ builder.config("spark.yarn.jars", yarnJars);
+ logChannel.logBasic(
+ "YARN containers get the engine's Spark runtime: "
+ + yarnJars.split(",").length
+ + " jars in spark.yarn.jars");
+ }
+ }
+ }
+
private void applySparkConfig(SparkSession.Builder builder, String key, String value) {
String resolved = resolve(value);
if (StringUtils.isNotEmpty(resolved)) {
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineRunConfiguration.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineRunConfiguration.java
index e3d478990c..8d1a6a81de 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineRunConfiguration.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineRunConfiguration.java
@@ -118,6 +118,15 @@ public class SparkPipelineRunConfiguration extends EmptyPipelineRunConfiguration
@HopMetadataProperty
private String fatJar;
+ @GuiWidgetElement(
+ order = "20025-spark-options",
+ parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.FOLDER,
+ label = "i18n::SparkEngine.OptionsHadoopConfDir.Label",
+ toolTip = "i18n::SparkEngine.OptionsHadoopConfDir.ToolTip")
+ @HopMetadataProperty
+ private String hadoopConfDir;
+
@GuiWidgetElement(
order = "20030-spark-options",
parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
@@ -213,6 +222,7 @@ public class SparkPipelineRunConfiguration extends EmptyPipelineRunConfiguration
this.sparkMaster = config.sparkMaster;
this.sparkAppName = config.sparkAppName;
this.fatJar = config.fatJar;
+ this.hadoopConfDir = config.hadoopConfDir;
this.sparkConfigs = config.sparkConfigs;
this.driverMemory = config.driverMemory;
this.executorMemory = config.executorMemory;
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/template/SparkRunConfigTemplate.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/template/SparkRunConfigTemplate.java
index f6bb7cfb1e..c8dcdec50b 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/template/SparkRunConfigTemplate.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/template/SparkRunConfigTemplate.java
@@ -85,7 +85,8 @@ spark.sql.shuffle.partitions=12
""),
YARN_CLIENT(
"YARN client",
- "Submit to YARN in client deploy mode (cluster provides Spark).",
+ "Run on YARN in client deploy mode. Set the Hadoop configuration folder (or HADOOP_CONF_DIR)"
+ + " and a native-provided fat jar.",
"yarn",
"Apache Hop",
"",
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
index 121949ea07..68265699d6 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
@@ -20,6 +20,7 @@ package org.apache.hop.spark.pipeline;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -59,6 +60,7 @@ import org.apache.hop.spark.util.SparkConst;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
+import org.apache.spark.storage.StorageLevel;
/**
* Converts a Hop {@link PipelineMeta} into a Spark {@link Dataset} lineage. Stateless transforms
@@ -113,6 +115,12 @@ public class HopPipelineMetaToSparkConverter {
private String parentLogChannelId;
private String dataSamplersJson;
+ /** Datasets of the transforms without successors, in topological order (see createDataset). */
+ private final Map<String, Dataset<Row>> leafDatasets = new LinkedHashMap<>();
+
+ /** Datasets read by more than one transform, persisted so they are computed once. */
+ private final List<Dataset<Row>> persistedDatasets = new ArrayList<>();
+
public HopPipelineMetaToSparkConverter(
IVariables variables,
PipelineMeta pipelineMeta,
@@ -201,7 +209,8 @@ public class HopPipelineMetaToSparkConverter {
/**
* Build the Dataset graph and return the sink Dataset (last transform with no successors, or the
- * last in topological order). Calling an action on the result materializes the pipeline.
+ * last in topological order). Generic transforms are lazy: run an action on every Dataset in
+ * {@link #getLeafDatasets()}, not only on the result, or the other branches never run.
*/
public Dataset<Row> createDataset(SparkSession spark) throws HopException {
try {
@@ -269,6 +278,18 @@ public class HopPipelineMetaToSparkConverter {
rowMeta,
previousTransforms,
input);
+
+ persistIfShared(log, transformMeta, transformDatasetMap);
+ }
+
+ leafDatasets.clear();
+ for (TransformMeta candidate : transforms) {
+ if (pipelineMeta.findNextTransforms(candidate).isEmpty()) {
+ Dataset<Row> leaf = transformDatasetMap.get(candidate.getName());
+ if (leaf != null) {
+ leafDatasets.put(candidate.getName(), leaf);
+ }
+ }
}
// Prefer a leaf transform as the result
@@ -296,6 +317,43 @@ public class HopPipelineMetaToSparkConverter {
}
}
+ /**
+ * Persists the output of a transform that more than one transform reads (target streams of Filter
+ * Rows / Switch / Case, a main and an info hop, ...). Without it every branch that runs an action
+ * recomputes the transform and everything before it.
+ */
+ private void persistIfShared(
+ ILogChannel log, TransformMeta transformMeta, Map<String, Dataset<Row>> transformDatasetMap) {
+ String name = transformMeta.getName();
+ Dataset<Row> shared = transformDatasetMap.get(HopSparkUtil.createSharedOutputId(name));
+ if (shared == null && pipelineMeta.findNextTransforms(transformMeta).size() > 1) {
+ shared = transformDatasetMap.get(name);
+ }
+ if (shared == null) {
+ return;
+ }
+ shared.persist(StorageLevel.MEMORY_AND_DISK());
+ persistedDatasets.add(shared);
+ log.logDetailed("Persisted the output of transform '" + name + "': several transforms read it");
+ }
+
+ /** Datasets of the transforms without successors, available after {@link #createDataset}. */
+ public Map<String, Dataset<Row>> getLeafDatasets() {
+ return leafDatasets;
+ }
+
+ /** Releases the Datasets persisted by {@link #createDataset}. */
+ public void unpersistAll() {
+ for (Dataset<Row> dataset : persistedDatasets) {
+ try {
+ dataset.unpersist();
+ } catch (Exception e) {
+ // The session may already be gone
+ }
+ }
+ persistedDatasets.clear();
+ }
+
/**
* Resolve Datasets for all main previous transforms (preferring target-stream keys), verify that
* Hop row layouts match, and {@code union} them. Matches Beam Flatten / Hop multi-hop merge.
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkGenericTransformHandler.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkGenericTransformHandler.java
index 27853707ba..b2436e217d 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkGenericTransformHandler.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkGenericTransformHandler.java
@@ -280,6 +280,8 @@ public class SparkGenericTransformHandler implements ISparkPipelineTransformHand
.equalTo(HopSparkUtil.MAIN_TARGET_TAG))
.drop(HopSparkUtil.TARGET_TAG_COLUMN);
transformDatasetMap.put(transformMeta.getName(), mainBranch);
+ transformDatasetMap.put(
+ HopSparkUtil.createSharedOutputId(transformMeta.getName()), taggedOrPlain);
for (String targetName : targetNames) {
Dataset<Row> branch =
taggedOrPlain
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/run/MainSpark.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/run/MainSpark.java
index 3b690c1cfd..2a14d4a260 100644
--- a/plugins/engines/spark/src/main/java/org/apache/hop/spark/run/MainSpark.java
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/run/MainSpark.java
@@ -17,8 +17,10 @@
package org.apache.hop.spark.run;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.file.Files;
import java.util.List;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
@@ -93,6 +95,21 @@ public class MainSpark {
}
}
+ /**
+ * The fat jar holds every plugin. Without this, Hop also scans {@code ./plugins}: started from a
+ * Hop installation folder, it then loads the Spark engine a second time with another class loader
+ * and the run configuration no longer matches its own class. A {@code -DHOP_PLUGIN_BASE_FOLDERS}
+ * system property is left alone.
+ */
+ private static void ignorePluginFoldersOutsideTheFatJar() throws IOException {
+ if (System.getProperty(Const.HOP_PLUGIN_BASE_FOLDERS) != null) {
+ return;
+ }
+ File noPlugins = Files.createTempDirectory("hop-spark-no-plugin-folders").toFile();
+ noPlugins.deleteOnExit();
+ System.setProperty(Const.HOP_PLUGIN_BASE_FOLDERS, noPlugins.getAbsolutePath());
+ }
+
private static void runPipeline(String[] args) throws Exception {
System.out.println(">>>>>> Initializing Hop");
System.out.println(
@@ -107,6 +124,7 @@ public class MainSpark {
} catch (Exception ignored) {
// best-effort diagnostics only
}
+ ignorePluginFoldersOutsideTheFatJar();
HopEnvironment.init();
MainSparkArgs parsed = MainSparkArgs.parse(args);
diff --git a/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkHadoopConfiguration.java b/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkHadoopConfiguration.java
new file mode 100644
index 0000000000..4f42fe6210
--- /dev/null
+++ b/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkHadoopConfiguration.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.spark.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.InputStream;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.UnaryOperator;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.vfs2.FileObject;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.vfs.HopVfs;
+
+/**
+ * Hadoop client configuration for the Spark sessions Hop starts itself (Hop GUI, hop-run, Hop
+ * Server). {@code spark-submit} puts {@code HADOOP_CONF_DIR} on the classpath; a Hop JVM does not,
+ * so the {@code *-site.xml} files are read here and passed to Spark as {@code spark.hadoop.*}.
+ */
+public final class SparkHadoopConfiguration {
+
+ public static final String ENV_HADOOP_CONF_DIR = "HADOOP_CONF_DIR";
+ public static final String ENV_YARN_CONF_DIR = "YARN_CONF_DIR";
+
+ /** Read in this order: later files win for keys they both set. */
+ public static final List<String> SITE_FILES =
+ List.of("core-site.xml", "hdfs-site.xml", "mapred-site.xml", "yarn-site.xml");
+
+ public static final String SPARK_HADOOP_PREFIX = "spark.hadoop.";
+
+ private SparkHadoopConfiguration() {}
+
+ /** Is this a YARN master URL ({@code yarn}, {@code yarn-client}, ...)? */
+ public static boolean isYarnMaster(String master) {
+ return master != null && master.trim().toLowerCase().startsWith("yarn");
+ }
+
+ /**
+ * The folders to read Hadoop configuration from. An explicit folder (run configuration) is used
+ * for every master. The {@code HADOOP_CONF_DIR} / {@code YARN_CONF_DIR} environment variables are
+ * only used for YARN: applied to {@code local[*]} runs they could point plain paths at HDFS.
+ *
+ * @param configuredFolder resolved folder from the run configuration, may be empty
+ * @param yarnMaster whether the session runs on YARN
+ * @param env environment lookup, {@code System::getenv} outside tests
+ * @return distinct folders, in reading order
+ */
+ public static List<String> resolveFolders(
+ String configuredFolder, boolean yarnMaster, UnaryOperator<String> env) {
+ List<String> folders = new ArrayList<>();
+ if (StringUtils.isNotBlank(configuredFolder)) {
+ folders.add(configuredFolder.trim());
+ return folders;
+ }
+ if (yarnMaster) {
+ for (String name : List.of(ENV_HADOOP_CONF_DIR, ENV_YARN_CONF_DIR)) {
+ String value = env.apply(name);
+ if (StringUtils.isNotBlank(value) && !folders.contains(value.trim())) {
+ folders.add(value.trim());
+ }
+ }
+ }
+ return folders;
+ }
+
+ /**
+ * Reads the site files found in the folders and returns their properties as {@code
+ * spark.hadoop.<key>} entries, with Hadoop variable expansion applied.
+ */
+ public static Map<String, String> load(List<String> folders, ILogChannel log)
+ throws HopException {
+ Configuration conf = new Configuration(false);
+ List<String> filesRead = new ArrayList<>();
+ for (String folder : folders) {
+ try {
+ FileObject dir = HopVfs.getFileObject(folder);
+ if (!dir.exists() || !dir.isFolder()) {
+ log.logError("Hadoop configuration folder '" + folder + "' does not exist; skipped");
+ continue;
+ }
+ for (String site : SITE_FILES) {
+ FileObject file = dir.resolveFile(site);
+ if (!file.exists()) {
+ continue;
+ }
+ byte[] bytes;
+ try (InputStream in = HopVfs.getInputStream(file)) {
+ bytes = IOUtils.toByteArray(in);
+ }
+ conf.addResource(new ByteArrayInputStream(bytes), file.getName().getURI());
+ filesRead.add(file.getName().getURI());
+ }
+ } catch (Exception e) {
+ throw new HopException("Error reading Hadoop configuration from '" + folder + "'", e);
+ }
+ }
+ Map<String, String> result = new LinkedHashMap<>();
+ if (filesRead.isEmpty()) {
+ return result;
+ }
+ for (Map.Entry<String, String> entry : conf) {
+ String value = conf.get(entry.getKey());
+ if (value != null) {
+ result.put(SPARK_HADOOP_PREFIX + entry.getKey(), value);
+ }
+ }
+ log.logBasic(
+ "Read "
+ + result.size()
+ + " Hadoop configuration properties from "
+ + String.join(", ", filesRead));
+ return result;
+ }
+
+ /**
+ * Classes Spark needs at runtime that the engine plugin's {@code lib} folder leaves to Hop's
+ * {@code lib/core} (see the plugin assembly): SLF4J and Jackson.
+ */
+ private static final List<String> RUNTIME_CLASSES_FROM_HOP =
+ List.of(
+ "org.slf4j.LoggerFactory",
+ "com.fasterxml.jackson.core.JsonFactory",
+ "com.fasterxml.jackson.databind.ObjectMapper",
+ "com.fasterxml.jackson.annotation.JsonProperty");
+
+ /**
+ * The {@code spark.yarn.jars} value that makes YARN containers run the Spark version Hop runs:
+ * the jars in the engine plugin's {@code lib} folder, plus the SLF4J and Jackson jars Hop itself
+ * uses.
+ *
+ * @param engineClass a class loaded from the engine plugin jar
+ * @return a comma-separated list of jar paths, or null when the plugin's {@code lib} folder
+ * cannot be found (for example in unit tests)
+ */
+ public static String defaultYarnJars(Class<?> engineClass) {
+ File lib;
+ try {
+ File pluginFolder = jarOf(engineClass).getParentFile();
+ if (pluginFolder == null) {
+ return null;
+ }
+ lib = new File(pluginFolder, "lib");
+ if (!lib.isDirectory()) {
+ return null;
+ }
+ } catch (Exception e) {
+ return null;
+ }
+ List<String> jars = new ArrayList<>();
+ File[] libJars = lib.listFiles((dir, name) -> name.endsWith(".jar"));
+ if (libJars == null) {
+ return null;
+ }
+ Arrays.sort(libJars);
+ for (File jar : libJars) {
+ // Hop classes reach the containers in the fat jar, on Spark's user class loader. A Hop jar
+ // here (lib/ also holds the lakehouse metadata jar) would sit on the containers' system
+ // class path, where it cannot see the Hop classes it extends.
+ if (!jar.getName().startsWith("hop-")) {
+ jars.add(jar.toPath().toAbsolutePath().normalize().toString());
+ }
+ }
+ ClassLoader classLoader = engineClass.getClassLoader();
+ for (String className : RUNTIME_CLASSES_FROM_HOP) {
+ try {
+ File jar = jarOf(Class.forName(className, false, classLoader));
+ String path = jar.toPath().toAbsolutePath().normalize().toString();
+ if (jar.isFile() && path.endsWith(".jar") && !jars.contains(path)) {
+ jars.add(path);
+ }
+ } catch (Exception e) {
+ // Not on the classpath: Spark on YARN reports what is missing
+ }
+ }
+ return String.join(",", jars);
+ }
+
+ private static File jarOf(Class<?> clazz) throws Exception {
+ URI location = clazz.getProtectionDomain().getCodeSource().getLocation().toURI();
+ return new File(location);
+ }
+}
diff --git a/plugins/engines/spark/src/main/resources/org/apache/hop/spark/engines/messages/messages_en_US.properties b/plugins/engines/spark/src/main/resources/org/apache/hop/spark/engines/messages/messages_en_US.properties
index 4483c854f3..6e9be2e896 100644
--- a/plugins/engines/spark/src/main/resources/org/apache/hop/spark/engines/messages/messages_en_US.properties
+++ b/plugins/engines/spark/src/main/resources/org/apache/hop/spark/engines/messages/messages_en_US.properties
@@ -28,6 +28,8 @@ SparkEngine.OptionsAppName.Label=Application name
SparkEngine.OptionsAppName.ToolTip=Spark application name
SparkEngine.OptionsFatJar.Label=Fat jar file location
SparkEngine.OptionsFatJar.ToolTip=Optional path to a Hop fat jar staged on executors for cluster runs. Leave empty for pure spark-submit with a native-provided fat jar.
+SparkEngine.OptionsHadoopConfDir.Label=Hadoop configuration folder
+SparkEngine.OptionsHadoopConfDir.ToolTip=Folder with core-site.xml, hdfs-site.xml and yarn-site.xml for runs that Hop starts itself (Hop GUI, hop-run, Hop Server). Empty: for a YARN master, '${HADOOP_CONF_DIR}' and '${YARN_CONF_DIR}' are used. Not used under spark-submit.
SparkEngine.OptionsConfigs.Label=Spark config (key=value lines)
SparkEngine.OptionsConfigs.ToolTip=Additional SparkConf entries, one key=value per line
SparkEngine.OptionsDriverMemory.Label=Driver memory
diff --git a/plugins/engines/spark/src/test/java/org/apache/hop/spark/engines/SparkPipelineRunConfigurationTest.java b/plugins/engines/spark/src/test/java/org/apache/hop/spark/engines/SparkPipelineRunConfigurationTest.java
index 2d68ac4395..b8d9d371c9 100644
--- a/plugins/engines/spark/src/test/java/org/apache/hop/spark/engines/SparkPipelineRunConfigurationTest.java
+++ b/plugins/engines/spark/src/test/java/org/apache/hop/spark/engines/SparkPipelineRunConfigurationTest.java
@@ -36,12 +36,14 @@ class SparkPipelineRunConfigurationTest {
config.setDriverMemory("1g");
config.setPathSchemeMap("s3=s3a\nminio=s3a");
config.setGenericTransformRunMode("DRIVER_ONLY");
+ config.setHadoopConfDir("/etc/hadoop/conf");
SparkPipelineRunConfiguration copy = config.clone();
assertNotSame(config, copy);
assertEquals("spark://host:7077", copy.getSparkMaster());
assertEquals("test-app", copy.getSparkAppName());
assertEquals("1g", copy.getDriverMemory());
+ assertEquals("/etc/hadoop/conf", copy.getHadoopConfDir());
assertEquals("s3=s3a\nminio=s3a", copy.getPathSchemeMap());
assertEquals("DRIVER_ONLY", copy.getGenericTransformRunMode());
assertEquals("DISTRIBUTED", new SparkPipelineRunConfiguration().getGenericTransformRunMode());
diff --git a/plugins/engines/spark/src/test/java/org/apache/hop/spark/util/SparkHadoopConfigurationTest.java b/plugins/engines/spark/src/test/java/org/apache/hop/spark/util/SparkHadoopConfigurationTest.java
new file mode 100644
index 0000000000..bcf2dd467e
--- /dev/null
+++ b/plugins/engines/spark/src/test/java/org/apache/hop/spark/util/SparkHadoopConfigurationTest.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.spark.util;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.logging.LogChannel;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class SparkHadoopConfigurationTest {
+
+ @TempDir Path tempDir;
+
+ @BeforeAll
+ static void init() throws Exception {
+ HopEnvironment.init();
+ }
+
+ private static String site(String... keyValues) {
+ StringBuilder xml = new StringBuilder("<?xml version=\"1.0\"?>\n<configuration>\n");
+ for (int i = 0; i < keyValues.length; i += 2) {
+ xml.append(" <property><name>")
+ .append(keyValues[i])
+ .append("</name><value>")
+ .append(keyValues[i + 1])
+ .append("</value></property>\n");
+ }
+ return xml.append("</configuration>\n").toString();
+ }
+
+ @Test
+ void configuredFolderIsUsedForEveryMaster() {
+ List<String> folders =
+ SparkHadoopConfiguration.resolveFolders("/etc/hadoop/conf", false, name -> "/other");
+ assertEquals(List.of("/etc/hadoop/conf"), folders);
+ }
+
+ @Test
+ void environmentIsOnlyUsedForYarn() {
+ Map<String, String> env =
+ Map.of("HADOOP_CONF_DIR", "/etc/hadoop/conf", "YARN_CONF_DIR", "/etc/hadoop/conf");
+ assertTrue(SparkHadoopConfiguration.resolveFolders("", false, env::get).isEmpty());
+ assertEquals(
+ List.of("/etc/hadoop/conf"), SparkHadoopConfiguration.resolveFolders("", true, env::get));
+ }
+
+ @Test
+ void yarnMasters() {
+ assertTrue(SparkHadoopConfiguration.isYarnMaster("yarn"));
+ assertTrue(SparkHadoopConfiguration.isYarnMaster(" YARN "));
+ assertFalse(SparkHadoopConfiguration.isYarnMaster("spark://host:7077"));
+ assertFalse(SparkHadoopConfiguration.isYarnMaster(null));
+ }
+
+ @Test
+ void siteFilesBecomeSparkHadoopProperties() throws Exception {
+ Files.writeString(
+ tempDir.resolve("core-site.xml"),
+ site("fs.defaultFS", "hdfs://namenode:8020", "hop.test.host", "namenode"));
+ Files.writeString(
+ tempDir.resolve("yarn-site.xml"),
+ site(
+ "yarn.resourcemanager.hostname",
+ "${hop.test.host}",
+ "yarn.resourcemanager.address",
+ "namenode:8050"));
+
+ Map<String, String> properties =
+ SparkHadoopConfiguration.load(
+ List.of(tempDir.toString()), new LogChannel("SparkHadoopConfigurationTest"));
+
+ assertEquals("hdfs://namenode:8020", properties.get("spark.hadoop.fs.defaultFS"));
+ assertEquals("namenode:8050", properties.get("spark.hadoop.yarn.resourcemanager.address"));
+ // Hadoop variable expansion is applied
+ assertEquals("namenode", properties.get("spark.hadoop.yarn.resourcemanager.hostname"));
+ }
+
+ @Test
+ void missingFolderOrFilesGiveNoProperties() throws Exception {
+ LogChannel log = new LogChannel("SparkHadoopConfigurationTest");
+ assertTrue(
+ SparkHadoopConfiguration.load(List.of(tempDir.resolve("nope").toString()), log).isEmpty());
+ assertTrue(SparkHadoopConfiguration.load(List.of(tempDir.toString()), log).isEmpty());
+ }
+
+ @Test
+ void noYarnJarsOutsideThePluginFolder() {
+ // Test classes are not loaded from plugins/engines/spark, so there is no lib folder next to
+ // them
+ assertNull(SparkHadoopConfiguration.defaultYarnJars(SparkHadoopConfigurationTest.class));
+ }
+}