This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new aeadc0692191 [SPARK-57978][PYTHON] Use explicit UTF-8 encoding in
text-mode I/O to prepare for PEP 686
aeadc0692191 is described below
commit aeadc0692191a07add6a499a2360c531a6b0e38e
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Jul 6 21:43:55 2026 -0700
[SPARK-57978][PYTHON] Use explicit UTF-8 encoding in text-mode I/O to
prepare for PEP 686
### What changes were proposed in this pull request?
This PR adds an explicit `encoding="utf-8"` to all text-mode I/O in the
PySpark runtime (non-test) code.
Note that
- The JVM <-> Python boundary is also unchanged: it already uses explicit
UTF-8 (`StandardCharsets.UTF_8` on the Scala side, binary sockets with
`UTF8Deserializer` on the Python side).
- the test code and doctests are unchanged (they read and write with the
same default, so they are self-consistent).
### Why are the changes needed?
[PEP 686](https://peps.python.org/pep-0686/) will make UTF-8 mode the
default in Python 3.15, changing the default encoding of text-mode I/O from the
locale encoding to UTF-8. On non-UTF-8 locales (Windows cp1252/cp949, POSIX
`C`), the same code would behave differently on Python <= 3.14 and 3.15.
Explicit encoding pins the behavior across all versions and locales.
### Does this PR introduce _any_ user-facing change?
No. On UTF-8 locales the behavior is identical; on non-UTF-8 locales, files
written by PySpark are now consistently UTF-8, matching the Python 3.15 default.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #57056 from dongjoon-hyun/SPARK-57978.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
python/packaging/classic/setup.py | 4 ++--
python/packaging/client/setup.py | 2 +-
python/packaging/connect/setup.py | 2 +-
python/pyspark/daemon.py | 6 ++++--
python/pyspark/errors/error_classes.py | 4 +++-
python/pyspark/errors/exceptions/__init__.py | 2 +-
python/pyspark/errors_doc_gen.py | 2 +-
python/pyspark/ml/connect/io_utils.py | 4 ++--
python/pyspark/ml/deepspeed/deepspeed_distributor.py | 4 +++-
python/pyspark/ml/dl_util.py | 2 +-
python/pyspark/ml/torch/distributor.py | 6 +++---
python/pyspark/ml/torch/torch_run_process_wrapper.py | 2 +-
python/pyspark/pandas/supported_api_gen.py | 2 +-
python/pyspark/pipelines/cli.py | 4 ++--
python/pyspark/pipelines/init_cli.py | 6 +++---
python/pyspark/profiler.py | 2 +-
python/pyspark/shell.py | 2 +-
python/pyspark/shuffle.py | 2 +-
python/pyspark/sql/profiler.py | 2 +-
python/pyspark/testing/sqlutils.py | 3 ++-
python/pyspark/util.py | 2 +-
21 files changed, 36 insertions(+), 29 deletions(-)
diff --git a/python/packaging/classic/setup.py
b/python/packaging/classic/setup.py
index 53d11a917f55..97ec9dd90ddb 100755
--- a/python/packaging/classic/setup.py
+++ b/python/packaging/classic/setup.py
@@ -45,7 +45,7 @@ if (
os.chdir(Path(__file__).parent.parent.parent.absolute())
try:
- exec(open("pyspark/version.py").read())
+ exec(open("pyspark/version.py", encoding="utf-8").read())
except IOError:
print(
"Failed to load PySpark version file for packaging. You must be in
Spark's python dir.",
@@ -258,7 +258,7 @@ try:
# will search for SPARK_HOME with Python.
scripts.append("pyspark/find_spark_home.py")
- with open("README.md") as f:
+ with open("README.md", encoding="utf-8") as f:
long_description = f.read()
setup(
diff --git a/python/packaging/client/setup.py b/python/packaging/client/setup.py
index 9c7df16d9537..6c97164937ee 100755
--- a/python/packaging/client/setup.py
+++ b/python/packaging/client/setup.py
@@ -142,7 +142,7 @@ try:
_minimum_pyyaml_version = "3.11"
_minimum_zstandard_version = "0.25.0"
- with open("README.md") as f:
+ with open("README.md", encoding="utf-8") as f:
long_description = f.read()
connect_packages = [
diff --git a/python/packaging/connect/setup.py
b/python/packaging/connect/setup.py
index 3df383f23a04..a4e83e41d151 100755
--- a/python/packaging/connect/setup.py
+++ b/python/packaging/connect/setup.py
@@ -95,7 +95,7 @@ try:
_minimum_pyyaml_version = "3.11"
_minimum_zstandard_version = "0.25.0"
- with open("README.md") as f:
+ with open("README.md", encoding="utf-8") as f:
long_description = f.read()
connect_packages = [
diff --git a/python/pyspark/daemon.py b/python/pyspark/daemon.py
index ef32bf760f47..0e321d442e81 100644
--- a/python/pyspark/daemon.py
+++ b/python/pyspark/daemon.py
@@ -98,7 +98,9 @@ def worker(sock: socket.socket, authenticated: bool) -> int:
faulthandler_log_path =
os.environ.get("PYTHON_FAULTHANDLER_DIR", None)
if faulthandler_log_path:
faulthandler_log_path =
os.path.join(faulthandler_log_path, str(os.getpid()))
- with open(faulthandler_log_path, "w") as
faulthandler_log_file:
+ with open(
+ faulthandler_log_path, "w", encoding="utf-8"
+ ) as faulthandler_log_file:
faulthandler.dump_traceback(file=faulthandler_log_file)
raise
else:
@@ -244,7 +246,7 @@ def manager() -> None:
# Therefore, here we redirects it to '/dev/null' by
duplicating
# another file descriptor for '/dev/null' to the
standard input (0).
# See SPARK-26175.
- devnull = open(os.devnull, "r")
+ devnull = open(os.devnull, "r", encoding="utf-8")
os.dup2(devnull.fileno(), 0)
devnull.close()
diff --git a/python/pyspark/errors/error_classes.py
b/python/pyspark/errors/error_classes.py
index 12094d61336e..4a840bafb8ae 100644
--- a/python/pyspark/errors/error_classes.py
+++ b/python/pyspark/errors/error_classes.py
@@ -23,6 +23,8 @@ import importlib.resources
# For more information, please see:
https://issues.apache.org/jira/browse/SPARK-46810
# This discrepancy will be resolved as part of:
https://issues.apache.org/jira/browse/SPARK-47429
ERROR_CLASSES_JSON = (
-
importlib.resources.files("pyspark.errors").joinpath("error-conditions.json").read_text()
+ importlib.resources.files("pyspark.errors")
+ .joinpath("error-conditions.json")
+ .read_text(encoding="utf-8")
)
ERROR_CLASSES_MAP = json.loads(ERROR_CLASSES_JSON)
diff --git a/python/pyspark/errors/exceptions/__init__.py
b/python/pyspark/errors/exceptions/__init__.py
index 259073164853..3de00c17500c 100644
--- a/python/pyspark/errors/exceptions/__init__.py
+++ b/python/pyspark/errors/exceptions/__init__.py
@@ -23,7 +23,7 @@ def _write_self() -> None:
ERRORS_DIR = Path(__file__).parents[1]
- with open(ERRORS_DIR / "error-conditions.json", "w") as f:
+ with open(ERRORS_DIR / "error-conditions.json", "w", encoding="utf-8") as
f:
json.dump(
error_classes.ERROR_CLASSES_MAP,
f,
diff --git a/python/pyspark/errors_doc_gen.py b/python/pyspark/errors_doc_gen.py
index 53b8b8d1e12f..c541cd8fb7e2 100644
--- a/python/pyspark/errors_doc_gen.py
+++ b/python/pyspark/errors_doc_gen.py
@@ -45,7 +45,7 @@ This is a list of common, named error classes returned by
PySpark which are defi
When writing PySpark errors, developers must use an error class from the list.
If an appropriate error class is not available, add a new one into the list.
For more information, please refer to `Contributing Error and Exception
<contributing.rst#contributing-error-and-exception>`_.
"""
- with open(output_rst_file_path, "w") as f:
+ with open(output_rst_file_path, "w", encoding="utf-8") as f:
f.write(header + "\n\n")
for error_key, error_details in ERROR_CLASSES_MAP.items():
f.write(error_key + "\n")
diff --git a/python/pyspark/ml/connect/io_utils.py
b/python/pyspark/ml/connect/io_utils.py
index fcc2e8d93c0f..f6108934f399 100644
--- a/python/pyspark/ml/connect/io_utils.py
+++ b/python/pyspark/ml/connect/io_utils.py
@@ -131,7 +131,7 @@ class ParamsReadWrite(Params):
def _save_to_local(self, path: str) -> None:
metadata = self._save_to_node_path(path, [])
- with open(os.path.join(path, _META_DATA_FILE_NAME), "w") as fp:
+ with open(os.path.join(path, _META_DATA_FILE_NAME), "w",
encoding="utf-8") as fp:
json.dump(metadata, fp)
def saveToLocal(self, path: str, *, overwrite: bool = False) -> None:
@@ -195,7 +195,7 @@ class ParamsReadWrite(Params):
@classmethod
def _load_from_local(cls, path: str) -> "Params":
- with open(os.path.join(path, _META_DATA_FILE_NAME), "r") as fp:
+ with open(os.path.join(path, _META_DATA_FILE_NAME), "r",
encoding="utf-8") as fp:
metadata = json.load(fp)
return cls._load_instance_from_metadata(metadata, path)
diff --git a/python/pyspark/ml/deepspeed/deepspeed_distributor.py
b/python/pyspark/ml/deepspeed/deepspeed_distributor.py
index 3fd1d3bb3246..162cd2adda52 100644
--- a/python/pyspark/ml/deepspeed/deepspeed_distributor.py
+++ b/python/pyspark/ml/deepspeed/deepspeed_distributor.py
@@ -100,7 +100,9 @@ class DeepspeedTorchDistributor(TorchDistributor):
@staticmethod
def _get_deepspeed_config_path(deepspeed_config: Union[str, Dict[str,
Any]]) -> str:
if isinstance(deepspeed_config, dict):
- with tempfile.NamedTemporaryFile(mode="w+", delete=False,
suffix=".json") as file:
+ with tempfile.NamedTemporaryFile(
+ mode="w+", delete=False, suffix=".json", encoding="utf-8"
+ ) as file:
json.dump(deepspeed_config, file)
return file.name
deepspeed_config_path = deepspeed_config
diff --git a/python/pyspark/ml/dl_util.py b/python/pyspark/ml/dl_util.py
index 26538c328353..89cf2ea8f43c 100644
--- a/python/pyspark/ml/dl_util.py
+++ b/python/pyspark/ml/dl_util.py
@@ -119,7 +119,7 @@ class FunctionPickler:
with open("{fn_output_path}", "wb") as f:
cloudpickle.dump(output, f)
""")
- with open(script_path, "w") as f:
+ with open(script_path, "w", encoding="utf-8") as f:
if prefix_code != "":
f.write(prefix_code)
f.write(code_snippet)
diff --git a/python/pyspark/ml/torch/distributor.py
b/python/pyspark/ml/torch/distributor.py
index dc0eda962312..67ba6adb03c4 100644
--- a/python/pyspark/ml/torch/distributor.py
+++ b/python/pyspark/ml/torch/distributor.py
@@ -482,7 +482,7 @@ class TorchDistributor(Distributor):
tail: collections.deque = collections.deque(maxlen=_TAIL_LINES_TO_KEEP)
try:
for line in task.stdout: # type: ignore
- decoded = line.decode()
+ decoded = line.decode("utf-8")
tail.append(decoded)
if redirect_to_stdout:
if (
@@ -883,7 +883,7 @@ class TorchDistributor(Distributor):
schema_file_path = os.path.join(save_dir, "schema.json")
schema_json_string = json.dumps(input_schema_json)
- with open(schema_file_path, "w") as f:
+ with open(schema_file_path, "w", encoding="utf-8") as f:
f.write(schema_json_string)
os.environ[SPARK_PARTITION_ARROW_DATA_FILE] = arrow_file_path
@@ -1088,7 +1088,7 @@ def _get_spark_partition_data_loader(
arrow_file = os.environ[SPARK_PARTITION_ARROW_DATA_FILE]
schema_file = os.environ[SPARK_DATAFRAME_SCHEMA_FILE]
- with open(schema_file, "r") as fp:
+ with open(schema_file, "r", encoding="utf-8") as fp:
schema = StructType.fromJson(json.load(fp))
dataset = _SparkPartitionTorchDataset(arrow_file, schema, num_samples)
diff --git a/python/pyspark/ml/torch/torch_run_process_wrapper.py
b/python/pyspark/ml/torch/torch_run_process_wrapper.py
index 67ec492329df..88415a2c8349 100644
--- a/python/pyspark/ml/torch/torch_run_process_wrapper.py
+++ b/python/pyspark/ml/torch/torch_run_process_wrapper.py
@@ -69,7 +69,7 @@ if __name__ == "__main__":
task.stdin.close() # type: ignore[union-attr]
try:
for line in task.stdout: # type: ignore[union-attr]
- decoded = line.decode()
+ decoded = line.decode("utf-8")
print(decoded.rstrip())
task.wait()
finally:
diff --git a/python/pyspark/pandas/supported_api_gen.py
b/python/pyspark/pandas/supported_api_gen.py
index c39b8880f14b..3a9c4f1d2468 100644
--- a/python/pyspark/pandas/supported_api_gen.py
+++ b/python/pyspark/pandas/supported_api_gen.py
@@ -419,7 +419,7 @@ def _write_rst(
all_supported_status : Dict
Collected support status data.
"""
- with open(output_rst_file_path, "w") as w_fd:
+ with open(output_rst_file_path, "w", encoding="utf-8") as w_fd:
w_fd.write(RST_HEADER)
for module_info, supported_status in all_supported_status.items():
module, module_path = module_info
diff --git a/python/pyspark/pipelines/cli.py b/python/pyspark/pipelines/cli.py
index d98e189611bd..9598987fc3f3 100644
--- a/python/pyspark/pipelines/cli.py
+++ b/python/pyspark/pipelines/cli.py
@@ -150,7 +150,7 @@ def find_pipeline_spec(current_dir: Path) -> Path:
def load_pipeline_spec(spec_path: Path) -> PipelineSpec:
"""Load the pipeline spec from a YAML file at the given path."""
- with spec_path.open("r") as f:
+ with spec_path.open("r", encoding="utf-8") as f:
return unpack_pipeline_spec(yaml.safe_load(f))
@@ -261,7 +261,7 @@ def register_definitions(
module_spec.loader.exec_module(module)
elif file.suffix == ".sql":
log_with_curr_timestamp(f"Registering SQL file
{file}...")
- with file.open("r") as f:
+ with file.open("r", encoding="utf-8") as f:
sql = f.read()
file_path_relative_to_spec = file.relative_to(path)
registry.register_sql(sql, file_path_relative_to_spec)
diff --git a/python/pyspark/pipelines/init_cli.py
b/python/pyspark/pipelines/init_cli.py
index 18bbb70ed9c1..291aa44018ec 100644
--- a/python/pyspark/pipelines/init_cli.py
+++ b/python/pyspark/pipelines/init_cli.py
@@ -58,7 +58,7 @@ def init(name: str) -> None:
# Write the spec file to the project directory
spec_file = project_dir / "spark-pipeline.yml"
- with open(spec_file, "w") as f:
+ with open(spec_file, "w", encoding="utf-8") as f:
spec_content = SPEC.replace("{{ name }}", name).replace("{{
storage_root }}", storage_path)
f.write(spec_content)
@@ -68,12 +68,12 @@ def init(name: str) -> None:
# Create the Python example file
python_example_file = transformations_dir /
"example_python_materialized_view.py"
- with open(python_example_file, "w") as f:
+ with open(python_example_file, "w", encoding="utf-8") as f:
f.write(PYTHON_EXAMPLE)
# Create the SQL example file
sql_example_file = transformations_dir /
"example_sql_materialized_view.sql"
- with open(sql_example_file, "w") as f:
+ with open(sql_example_file, "w", encoding="utf-8") as f:
f.write(SQL_EXAMPLE)
print(f"Pipeline project '{name}' created successfully. To run your
pipeline:")
diff --git a/python/pyspark/profiler.py b/python/pyspark/profiler.py
index 8f797b5f21fc..45ab1e5afdbf 100644
--- a/python/pyspark/profiler.py
+++ b/python/pyspark/profiler.py
@@ -410,7 +410,7 @@ class MemoryProfiler(Profiler):
stats = self.stats() # dict
if stats:
p = os.path.join(path, "udf_%d_memory.txt" % id)
- with open(p, "w+") as f:
+ with open(p, "w+", encoding="utf-8") as f:
self._show_results(stats, stream=f)
diff --git a/python/pyspark/shell.py b/python/pyspark/shell.py
index dae4854a237e..435cc9de3b2e 100644
--- a/python/pyspark/shell.py
+++ b/python/pyspark/shell.py
@@ -138,6 +138,6 @@ print("SparkSession available as 'spark'.")
# which allows us to execute the user's PYTHONSTARTUP file:
_pythonstartup = os.environ.get("OLD_PYTHONSTARTUP")
if _pythonstartup and os.path.isfile(_pythonstartup):
- with open(_pythonstartup) as f:
+ with open(_pythonstartup, encoding="utf-8") as f:
code = compile(f.read(), _pythonstartup, "exec")
exec(code)
diff --git a/python/pyspark/shuffle.py b/python/pyspark/shuffle.py
index bfdc0b79ce41..c37bb7fbce30 100644
--- a/python/pyspark/shuffle.py
+++ b/python/pyspark/shuffle.py
@@ -80,7 +80,7 @@ def get_used_memory() -> int:
else:
if platform.system() == "Linux":
- for line in open("/proc/self/status"):
+ for line in open("/proc/self/status", encoding="utf-8"):
if line.startswith("VmRSS:"):
return int(line.split()[1]) >> 10
diff --git a/python/pyspark/sql/profiler.py b/python/pyspark/sql/profiler.py
index 60f68c61708f..eae9b44f449b 100644
--- a/python/pyspark/sql/profiler.py
+++ b/python/pyspark/sql/profiler.py
@@ -394,7 +394,7 @@ class ProfilerCollector(ABC):
os.makedirs(path, exist_ok=True)
p = os.path.join(path, f"udf_{id}_memory.txt")
- with open(p, "w+") as f:
+ with open(p, "w+", encoding="utf-8") as f:
MemoryProfiler._show_results(cm, stream=f)
if id is not None:
diff --git a/python/pyspark/testing/sqlutils.py
b/python/pyspark/testing/sqlutils.py
index e9d99dd3d2d6..1760ad5ffd79 100644
--- a/python/pyspark/testing/sqlutils.py
+++ b/python/pyspark/testing/sqlutils.py
@@ -84,6 +84,7 @@ def get_sbt_runtime_classpath(project_relative_path,
project_name_map):
cwd=SPARK_HOME,
capture_output=True,
text=True,
+ encoding="utf-8",
timeout=180,
)
@@ -138,7 +139,7 @@ def read_classpath(project_relative_path,
project_name_map=None):
# First try to read classpath.txt (Maven builds)
if os.path.exists(classpath_file):
- with open(classpath_file, "r") as f:
+ with open(classpath_file, "r", encoding="utf-8") as f:
classpath = f.read().strip()
# Replace colon with comma for spark-submit --jars format
return classpath.replace(":", ",")
diff --git a/python/pyspark/util.py b/python/pyspark/util.py
index 0d1440962f7f..858dc1013e94 100644
--- a/python/pyspark/util.py
+++ b/python/pyspark/util.py
@@ -978,7 +978,7 @@ class _FaulthandlerHelper:
self._log_path = os.environ.get("PYTHON_FAULTHANDLER_DIR", None)
if self._log_path:
self._log_path = os.path.join(self._log_path, str(os.getpid()))
- self._log_file = open(self._log_path, "w")
+ self._log_file = open(self._log_path, "w", encoding="utf-8")
faulthandler.enable(file=self._log_file)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
