This is an automated email from the ASF dual-hosted git repository.
mattcasters 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 f260c7e484 Issue #8946 : Write local Spark files on Windows without
winutils (#8951)
f260c7e484 is described below
commit f260c7e484a0fa454a35dc5fb39c2a1501a8f45d
Author: Matt Casters <[email protected]>
AuthorDate: Sat Oct 24 11:34:44 2026 +0200
Issue #8946 : Write local Spark files on Windows without winutils (#8951)
Hadoop's local file system shells out to winutils.exe when Spark file
output creates the output directory. On Windows that fails with "HADOOP_HOME
and hadoop.home.dir are unset" when neither is set.
When Hop creates the Spark session and winutils is missing, skip those
permission changes. An explicit spark.hadoop.fs.file.impl in the run
configuration still wins.
---
.../pipeline/spark/paths-and-filesystems.adoc | 11 ++
.../pipeline/transforms/spark-file-output.adoc | 1 +
.../hop/spark/engines/SparkPipelineEngine.java | 6 +
.../hop/spark/fs/HopWindowsLocalFileSystem.java | 145 +++++++++++++++++++++
.../spark/fs/HopWindowsLocalFileSystemTest.java | 88 +++++++++++++
5 files changed, 251 insertions(+)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/paths-and-filesystems.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/paths-and-filesystems.adoc
index e25db56193..d5107214d3 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/paths-and-filesystems.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/paths-and-filesystems.adoc
@@ -78,6 +78,17 @@ For the full list of Hop VFS schemes, see
xref:vfs.adoc[Virtual File Systems].
|Driver must open the definition files; *data* paths still must work on
executors
|===
+[[windows-local-paths]]
+== Windows and local mode
+
+On Windows, Hadoop's local file system sets POSIX permissions by running
`winutils.exe` from `HADOOP_HOME` (or the `hadoop.home.dir` system property).
Spark file output hits that call when the commit creates the output directory.
With neither variable set, the write fails with `HADOOP_HOME and
hadoop.home.dir are unset`.
+
+When Hop creates the Spark session on Windows and `winutils.exe` is not
installed, local `file:///` reads and writes skip those permission changes. The
files keep the permissions of the Hop process. You do not need a Hadoop
installation for `local[*]`.
+
+If `HADOOP_HOME` points at a directory that contains `bin/winutils.exe`, Hop
keeps Hadoop's own local file system and permissions are applied. To choose a
file system yourself, set `spark.hadoop.fs.file.impl` in the run configuration.
That entry wins.
+
+`hdfs://` and `s3a://` paths are unchanged. They do not use winutils.
+
== Scheme collision matrix
[options="header",cols="1,2,2,2"]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-file-output.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-file-output.adoc
index f2e6fd7962..d070eca710 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-file-output.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-file-output.adoc
@@ -31,6 +31,7 @@ This transform is **not** available on the local Hop engine
or on the Beam engin
It is a leaf: the write is executed as a Spark action when the engine
materialises the pipeline, and no rows are passed on.
The path is a Spark/Hadoop `FileSystem` URI that must be writable from every
executor, **not** a Hop VFS path: a local directory on `local[*]`, `hdfs://…`
or `s3a://…` on a cluster.
+On Windows, a `local[*]` path does not require `HADOOP_HOME` or
`winutils.exe`. See
xref:pipeline/spark/paths-and-filesystems.adoc#windows-local-paths[Windows and
local mode].
Spark always writes a **directory** of part files at that location, never a
single file with that name; see <<single-file>> below.
See xref:pipeline/spark/paths-and-filesystems.adoc[Paths and file systems] for
the scheme mapping and for the *Path scheme map* on the run configuration.
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 86bb346886..7defa9fc68 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
@@ -90,6 +90,7 @@ import org.apache.hop.spark.core.SparkNativeMetricsListener;
import org.apache.hop.spark.core.SparkTransformMetricSlice;
import org.apache.hop.spark.core.SparkTransformMetricsAccumulator;
import org.apache.hop.spark.execution.SparkTransformExecutionSampling;
+import org.apache.hop.spark.fs.HopWindowsLocalFileSystem;
import org.apache.hop.spark.pipeline.HopPipelineMetaToSparkConverter;
import org.apache.hop.spark.pkg.SparkProjectPackage;
import org.apache.hop.spark.table.LakeSessionPlan;
@@ -1135,6 +1136,11 @@ public class SparkPipelineEngine extends Variables
implements IPipelineEngine<Pi
applyHadoopConfiguration(builder, master);
}
+ // Windows without winutils fails in FileOutputCommitter.setupJob when
Hadoop
+ // chmods the output directory. Apply before run-config extras so an
explicit
+ // spark.hadoop.fs.file.impl replaces this default.
+ HopWindowsLocalFileSystem.applyIfNeeded(builder, logChannel);
+
Set<String> configuredKeys = new HashSet<>();
String extra = sparkEngineRunConfiguration.getSparkConfigs();
if (StringUtils.isNotEmpty(extra)) {
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystem.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystem.java
new file mode 100644
index 0000000000..0cf0a2c319
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystem.java
@@ -0,0 +1,145 @@
+/*
+ * 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.fs;
+
+import java.io.IOException;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.nativeio.NativeIO;
+import org.apache.hadoop.util.Shell;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Local Hadoop file system for a Windows Spark session that has no {@code
winutils.exe}.
+ *
+ * <p>Spark file output commits through {@code FileOutputCommitter}, which
creates the output
+ * directory with {@code FileSystem.mkdirs}. On Windows, Hadoop's {@code
RawLocalFileSystem} then
+ * runs {@code winutils chmod}. That binary is resolved from {@code
HADOOP_HOME} or {@code
+ * hadoop.home.dir}, and the call fails with {@code HADOOP_HOME and
hadoop.home.dir are unset} when
+ * neither is set. The file itself was already created by Java; only the
permission change fails.
+ *
+ * <p>This class keeps the checksummed local file system and skips those
permission changes. Files
+ * keep the permissions of the Hop process. It is installed only when Hop
creates the session, and
+ * only when Windows has neither winutils nor Hadoop's native I/O library. An
explicit {@code
+ * spark.hadoop.fs.file.impl} in the run configuration is applied afterwards
and wins.
+ */
+public class HopWindowsLocalFileSystem extends LocalFileSystem {
+
+ /** Spark conf key copied into Hadoop {@code fs.file.impl}. */
+ public static final String SPARK_HADOOP_FILE_IMPL =
"spark.hadoop.fs.file.impl";
+
+ public HopWindowsLocalFileSystem() {
+ super(new WinutilsFreeRawLocalFileSystem());
+ }
+
+ /**
+ * True when a local Spark write would fail in {@code
RawLocalFileSystem.setPermission} because
+ * Windows has no winutils binary and no Hadoop native library.
+ */
+ public static boolean needed() {
+ if (!Shell.WINDOWS) {
+ return false;
+ }
+ return permissionShellMissing(true, Shell.hasWinutilsPath(),
NativeIO.isAvailable());
+ }
+
+ /**
+ * Whether a permission change would shell out to a missing winutils binary.
+ *
+ * @param windows true when the JVM is running on Windows
+ * @param winutilsPresent true when HADOOP_HOME/bin/winutils.exe was found
+ * @param nativeIo true when Hadoop's native library can chmod without
winutils
+ * @return true when Windows has neither winutils nor the native library
+ */
+ public static boolean permissionShellMissing(
+ boolean windows, boolean winutilsPresent, boolean nativeIo) {
+ return windows && !winutilsPresent && !nativeIo;
+ }
+
+ /**
+ * Point a new Spark session at this file system when {@link #needed()} is
true.
+ *
+ * <p>Call this before run-configuration entries are copied onto the
builder. A later {@code
+ * spark.hadoop.fs.file.impl} replaces this default.
+ */
+ public static void applyIfNeeded(SparkSession.Builder builder, ILogChannel
log) {
+ if (!needed()) {
+ return;
+ }
+ builder.config(SPARK_HADOOP_FILE_IMPL,
HopWindowsLocalFileSystem.class.getName());
+ if (log != null) {
+ log.logBasic(
+ "HADOOP_HOME and hadoop.home.dir are unset. Local Spark paths will
keep the"
+ + " permissions of this process instead of calling winutils.");
+ }
+ }
+
+ /**
+ * Raw local file system that does not shell out for chmod or for owner
lookups.
+ *
+ * <p>{@code mkdirs} and {@code create} call {@code setPermission} on this
instance, not on the
+ * checksum wrapper, so the override has to live here.
+ */
+ public static class WinutilsFreeRawLocalFileSystem extends
RawLocalFileSystem {
+
+ @Override
+ public void setPermission(Path path, FsPermission permission) throws
IOException {
+ if (HopWindowsLocalFileSystem.needed()) {
+ return;
+ }
+ super.setPermission(path, permission);
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return localPermissions(super.getFileStatus(path));
+ }
+
+ @Override
+ public FileStatus getFileLinkStatus(Path path) throws IOException {
+ return localPermissions(super.getFileLinkStatus(path));
+ }
+
+ /**
+ * Copy status onto a plain {@link FileStatus}. The deprecated Windows
status loads owner and
+ * mode by running winutils the first time those fields are read.
+ */
+ private static FileStatus localPermissions(FileStatus status) throws
IOException {
+ if (!HopWindowsLocalFileSystem.needed()) {
+ return status;
+ }
+ Path symlink = status.isSymlink() ? status.getSymlink() : null;
+ return new FileStatus(
+ status.getLen(),
+ status.isDirectory(),
+ status.getReplication(),
+ status.getBlockSize(),
+ status.getModificationTime(),
+ status.getAccessTime(),
+ FsPermission.getFileDefault(),
+ "",
+ "",
+ symlink,
+ status.getPath());
+ }
+ }
+}
diff --git
a/plugins/engines/spark/src/test/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystemTest.java
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystemTest.java
new file mode 100644
index 0000000000..43d51a58ca
--- /dev/null
+++
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/fs/HopWindowsLocalFileSystemTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.fs;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.util.Shell;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class HopWindowsLocalFileSystemTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void permissionShellIsMissingOnlyOnWindowsWithoutWinutilsOrNativeIo() {
+ assertFalse(HopWindowsLocalFileSystem.permissionShellMissing(false, false,
false));
+ assertFalse(HopWindowsLocalFileSystem.permissionShellMissing(false, true,
true));
+ assertFalse(HopWindowsLocalFileSystem.permissionShellMissing(true, true,
false));
+ assertFalse(HopWindowsLocalFileSystem.permissionShellMissing(true, false,
true));
+ assertTrue(HopWindowsLocalFileSystem.permissionShellMissing(true, false,
false));
+ if (!Shell.WINDOWS) {
+ assertFalse(HopWindowsLocalFileSystem.needed());
+ }
+ }
+
+ @Test
+ void createsADirectoryAndFileWithoutASeparatePermissionTool() throws
Exception {
+ HopWindowsLocalFileSystem fs = new HopWindowsLocalFileSystem();
+ try {
+ fs.initialize(URI.create("file:///"), new Configuration());
+
assertInstanceOf(HopWindowsLocalFileSystem.WinutilsFreeRawLocalFileSystem.class,
fs.getRaw());
+
+ Path dir = new Path(tempDir.resolve("spark-out").toUri());
+ assertTrue(fs.mkdirs(dir));
+ Path file = new Path(dir, "part.txt");
+ try (FSDataOutputStream out = fs.create(file, true)) {
+ out.write("hello".getBytes(StandardCharsets.UTF_8));
+ }
+ fs.setPermission(file, FsPermission.getFileDefault());
+
+ assertTrue(fs.exists(file));
+ FileStatus status = fs.getFileStatus(file);
+ assertEquals(5, status.getLen());
+ assertEquals("hello", read(fs, file));
+ if (HopWindowsLocalFileSystem.needed()) {
+ assertEquals("", status.getOwner());
+ assertEquals(FsPermission.getFileDefault(), status.getPermission());
+ }
+ } finally {
+ fs.close();
+ }
+ }
+
+ private static String read(HopWindowsLocalFileSystem fs, Path file) throws
Exception {
+ byte[] buf = new byte[5];
+ try (FSDataInputStream in = fs.open(file)) {
+ assertEquals(5, in.read(buf));
+ }
+ return new String(buf, StandardCharsets.UTF_8);
+ }
+}
