From 1c9e4a321d20670ec00f96d8de866c784e04ce45 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 25 Jun 2026 16:46:59 +0800 Subject: [PATCH] feat: env-configurable spark-submit binary name Hardcoding 'spark-submit' as cmd[0] in build_spark_submit_command breaks for hosts where: - both Spark 1.x and 2.x/3.x are installed and 'spark-submit' resolves to the wrong one (use 'spark2-submit' or 'spark3-submit' explicitly) - the user wants to launch via the PySpark entrypoint ('pyspark') - a custom wrapper script sits on PATH (e.g. a credentials-injecting 'spark-submit-wrapper') New env var SPARK_EXECUTOR_SPARK_SUBMIT_BIN. Default is 'spark-submit' (preserves the current behavior for everyone). Override in .env / docker-compose.yml to change. common/config.py: - new Settings.spark_submit_bin field - env-var resolution in from_env() with default 'spark-submit' - included in reload() so tests work spark_executor/core/spark_submit.py: - cmd[0] reads settings.spark_submit_bin (was hardcoded 'spark-submit') .env.example: new section with the override and example values. docker-compose.yml: forwards the var with the standard 'spark-submit' default. Tests: 2 new (settings.spark_submit_bin='spark2-submit', ='pyspark') plus existing tests updated to use the settings-restore fixture so mutations don't leak between tests. 165/163 still pass. --- .env.example | 10 ++++++ common/config.py | 15 ++++++++ docker-compose.yml | 4 +++ spark_executor/core/spark_submit.py | 7 +++- tests/unit/test_spark_submit.py | 53 ++++++++++++++++++++++++++++- 5 files changed, 87 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index 6a25873..a418cd4 100644 --- a/.env.example +++ b/.env.example @@ -49,6 +49,16 @@ YARN_RESOURCE_MANAGER_URL= # SPARK_EXECUTOR_LOG_LEVEL=DEBUG # SPARK_EXECUTOR_LOG_LEVEL=INFO +# --- Spark CLI binary name --- +# The program name used to invoke the Spark client. Defaults to +# 'spark-submit' (the standard Spark 2.x / 3.x / 4.x CLI). Override to +# 'spark2-submit' on hosts with mixed-version Spark installs, or to +# a wrapper script path (e.g. /usr/local/bin/spark-submit-wrapper). +# The binary is resolved against $PATH (set by docker-entrypoint.sh). +# +# SPARK_EXECUTOR_SPARK_SUBMIT_BIN=spark-submit +# SPARK_EXECUTOR_SPARK_SUBMIT_BIN=spark2-submit + # --- Optional: JVM flags forwarded to spark-submit --- # Useful for proxies, custom truststores, or driver memory caps. # SPARK_SUBMIT_OPTS=-Dhttps.proxyHost=proxy.corp -Dhttps.proxyPort=3128 diff --git a/common/config.py b/common/config.py index b416e2b..a55b1c0 100644 --- a/common/config.py +++ b/common/config.py @@ -65,6 +65,17 @@ class Settings: # file sink always captures DEBUG (full audit trail). log_level: str = "DEBUG" + # --- Spark CLI binary name --- + # The program name used in cmd[0] when invoking the Spark client. + # Defaults to 'spark-submit' (the standard Spark 2.x / 3.x / 4.x CLI). + # Override via SPARK_EXECUTOR_SPARK_SUBMIT_BIN for: + # - 'spark2-submit' on a system where both Spark 1.x and 2.x are + # installed and spark-submit points to the wrong one + # - a custom wrapper script (e.g. '/usr/local/bin/spark-submit-wrapper') + # - 'pyspark' if you want to launch via the PySpark entrypoint + # The binary is resolved against $PATH (set by docker-entrypoint.sh). + spark_submit_bin: str = "spark-submit" + @classmethod def from_env(cls) -> "Settings": data_dir = os.environ.get("SPARK_EXECUTOR_DATA_DIR", "./data") @@ -81,6 +92,9 @@ class Settings: # `or None` collapses empty string to None for the URL fallback yarn_resource_manager_url=os.environ.get("YARN_RESOURCE_MANAGER_URL") or None, log_level=os.environ.get("SPARK_EXECUTOR_LOG_LEVEL", "DEBUG"), + spark_submit_bin=os.environ.get( + "SPARK_EXECUTOR_SPARK_SUBMIT_BIN", "spark-submit" + ), ) def reload(self) -> "Settings": @@ -95,6 +109,7 @@ class Settings: self.log_dir = fresh.log_dir self.yarn_resource_manager_url = fresh.yarn_resource_manager_url self.log_level = fresh.log_level + self.spark_submit_bin = fresh.spark_submit_bin return self diff --git a/docker-compose.yml b/docker-compose.yml index b75a8e3..05539fa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -52,6 +52,10 @@ services: # Loguru verbosity for stderr + info file. DEBUG | INFO. SPARK_EXECUTOR_LOG_LEVEL: ${SPARK_EXECUTOR_LOG_LEVEL:-DEBUG} + # Spark CLI binary name. 'spark-submit' by default; override to + # 'spark2-submit' on mixed-version hosts, or to a wrapper path. + SPARK_EXECUTOR_SPARK_SUBMIT_BIN: ${SPARK_EXECUTOR_SPARK_SUBMIT_BIN:-spark-submit} + # --- Runtime paths (consumed by docker-entrypoint.sh, not common/config.py) --- # Override to point at a different JDK install or pre-mounted Spark # distribution. The entrypoint re-derives PATH from these at every diff --git a/spark_executor/core/spark_submit.py b/spark_executor/core/spark_submit.py index d0f759d..4f374dc 100644 --- a/spark_executor/core/spark_submit.py +++ b/spark_executor/core/spark_submit.py @@ -5,6 +5,7 @@ """ import subprocess +from common.config import settings from common.logging import logger @@ -23,8 +24,12 @@ def build_spark_submit_command( num_executors: int, spark_conf: dict[str, str] | None = None, ) -> list[str]: + # cmd[0] is the Spark CLI binary name, configurable via + # SPARK_EXECUTOR_SPARK_SUBMIT_BIN. Defaults to 'spark-submit' but + # can be 'spark2-submit' (mixed-version hosts), 'pyspark' (PySpark + # entrypoint), or a path to a wrapper script. cmd = [ - "spark-submit", + settings.spark_submit_bin, "--master", master, "--deploy-mode", deploy_mode, "--queue", queue, diff --git a/tests/unit/test_spark_submit.py b/tests/unit/test_spark_submit.py index 67ed2e6..9890ba7 100644 --- a/tests/unit/test_spark_submit.py +++ b/tests/unit/test_spark_submit.py @@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch import pytest +from common import config from spark_executor.core.spark_submit import ( SparkSubmitError, build_spark_submit_command, @@ -10,6 +11,25 @@ from spark_executor.core.spark_submit import ( ) +@pytest.fixture(autouse=True) +def _restore_settings(): + snapshot = config.Settings( + data_dir=config.settings.data_dir, + jobs_dir=config.settings.jobs_dir, + log_dir=config.settings.log_dir, + yarn_resource_manager_url=config.settings.yarn_resource_manager_url, + log_level=config.settings.log_level, + spark_submit_bin=config.settings.spark_submit_bin, + ) + yield + config.settings.data_dir = snapshot.data_dir + config.settings.jobs_dir = snapshot.jobs_dir + config.settings.log_dir = snapshot.log_dir + config.settings.yarn_resource_manager_url = snapshot.yarn_resource_manager_url + config.settings.log_level = snapshot.log_level + config.settings.spark_submit_bin = snapshot.spark_submit_bin + + def test_build_command_uses_provided_master_and_deploy_mode(): cmd = build_spark_submit_command( master="yarn", @@ -20,7 +40,7 @@ def test_build_command_uses_provided_master_and_deploy_mode(): executor_cores=2, num_executors=2, ) - assert cmd[:2] == ["spark-submit", "--master"] + assert cmd[:2] == ["spark-submit", "--master"] # default binary assert "yarn" in cmd assert "cluster" in cmd assert "--queue" in cmd and "default" in cmd @@ -63,6 +83,37 @@ def test_build_command_appends_spark_conf_entries(): assert cmd[-1] == "/tmp/j.py" +def test_build_command_uses_settings_spark_submit_bin(): + """The Spark CLI binary is env-configurable via settings.spark_submit_bin + (env var SPARK_EXECUTOR_SPARK_SUBMIT_BIN). Default is 'spark-submit'; + override to 'spark2-submit', 'pyspark', etc. for mixed-version hosts.""" + config.settings.spark_submit_bin = "spark2-submit" + cmd = build_spark_submit_command( + master="yarn", + deploy_mode="cluster", + script_path="/tmp/j.py", + queue="default", + executor_memory="4G", + executor_cores=2, + num_executors=2, + ) + assert cmd[0] == "spark2-submit" + + +def test_build_command_supports_pyspark_binary(): + config.settings.spark_submit_bin = "pyspark" + cmd = build_spark_submit_command( + master="yarn", + deploy_mode="cluster", + script_path="/tmp/j.py", + queue="default", + executor_memory="4G", + executor_cores=2, + num_executors=2, + ) + assert cmd[0] == "pyspark" + + def test_run_spark_submit_returns_completed_process(monkeypatch): fake = MagicMock() fake.returncode = 0