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