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 <[email protected]>
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));
+ }
+}
