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.
This commit is contained in:
@@ -49,6 +49,16 @@ YARN_RESOURCE_MANAGER_URL=
|
|||||||
# SPARK_EXECUTOR_LOG_LEVEL=DEBUG
|
# SPARK_EXECUTOR_LOG_LEVEL=DEBUG
|
||||||
# SPARK_EXECUTOR_LOG_LEVEL=INFO
|
# 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 ---
|
# --- Optional: JVM flags forwarded to spark-submit ---
|
||||||
# Useful for proxies, custom truststores, or driver memory caps.
|
# Useful for proxies, custom truststores, or driver memory caps.
|
||||||
# SPARK_SUBMIT_OPTS=-Dhttps.proxyHost=proxy.corp -Dhttps.proxyPort=3128
|
# SPARK_SUBMIT_OPTS=-Dhttps.proxyHost=proxy.corp -Dhttps.proxyPort=3128
|
||||||
|
|||||||
@@ -65,6 +65,17 @@ class Settings:
|
|||||||
# file sink always captures DEBUG (full audit trail).
|
# file sink always captures DEBUG (full audit trail).
|
||||||
log_level: str = "DEBUG"
|
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
|
@classmethod
|
||||||
def from_env(cls) -> "Settings":
|
def from_env(cls) -> "Settings":
|
||||||
data_dir = os.environ.get("SPARK_EXECUTOR_DATA_DIR", "./data")
|
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
|
# `or None` collapses empty string to None for the URL fallback
|
||||||
yarn_resource_manager_url=os.environ.get("YARN_RESOURCE_MANAGER_URL") or None,
|
yarn_resource_manager_url=os.environ.get("YARN_RESOURCE_MANAGER_URL") or None,
|
||||||
log_level=os.environ.get("SPARK_EXECUTOR_LOG_LEVEL", "DEBUG"),
|
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":
|
def reload(self) -> "Settings":
|
||||||
@@ -95,6 +109,7 @@ class Settings:
|
|||||||
self.log_dir = fresh.log_dir
|
self.log_dir = fresh.log_dir
|
||||||
self.yarn_resource_manager_url = fresh.yarn_resource_manager_url
|
self.yarn_resource_manager_url = fresh.yarn_resource_manager_url
|
||||||
self.log_level = fresh.log_level
|
self.log_level = fresh.log_level
|
||||||
|
self.spark_submit_bin = fresh.spark_submit_bin
|
||||||
return self
|
return self
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -52,6 +52,10 @@ services:
|
|||||||
# Loguru verbosity for stderr + info file. DEBUG | INFO.
|
# Loguru verbosity for stderr + info file. DEBUG | INFO.
|
||||||
SPARK_EXECUTOR_LOG_LEVEL: ${SPARK_EXECUTOR_LOG_LEVEL:-DEBUG}
|
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) ---
|
# --- Runtime paths (consumed by docker-entrypoint.sh, not common/config.py) ---
|
||||||
# Override to point at a different JDK install or pre-mounted Spark
|
# Override to point at a different JDK install or pre-mounted Spark
|
||||||
# distribution. The entrypoint re-derives PATH from these at every
|
# distribution. The entrypoint re-derives PATH from these at every
|
||||||
|
|||||||
@@ -5,6 +5,7 @@
|
|||||||
"""
|
"""
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
|
from common.config import settings
|
||||||
from common.logging import logger
|
from common.logging import logger
|
||||||
|
|
||||||
|
|
||||||
@@ -23,8 +24,12 @@ def build_spark_submit_command(
|
|||||||
num_executors: int,
|
num_executors: int,
|
||||||
spark_conf: dict[str, str] | None = None,
|
spark_conf: dict[str, str] | None = None,
|
||||||
) -> list[str]:
|
) -> 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 = [
|
cmd = [
|
||||||
"spark-submit",
|
settings.spark_submit_bin,
|
||||||
"--master", master,
|
"--master", master,
|
||||||
"--deploy-mode", deploy_mode,
|
"--deploy-mode", deploy_mode,
|
||||||
"--queue", queue,
|
"--queue", queue,
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from common import config
|
||||||
from spark_executor.core.spark_submit import (
|
from spark_executor.core.spark_submit import (
|
||||||
SparkSubmitError,
|
SparkSubmitError,
|
||||||
build_spark_submit_command,
|
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():
|
def test_build_command_uses_provided_master_and_deploy_mode():
|
||||||
cmd = build_spark_submit_command(
|
cmd = build_spark_submit_command(
|
||||||
master="yarn",
|
master="yarn",
|
||||||
@@ -20,7 +40,7 @@ def test_build_command_uses_provided_master_and_deploy_mode():
|
|||||||
executor_cores=2,
|
executor_cores=2,
|
||||||
num_executors=2,
|
num_executors=2,
|
||||||
)
|
)
|
||||||
assert cmd[:2] == ["spark-submit", "--master"]
|
assert cmd[:2] == ["spark-submit", "--master"] # default binary
|
||||||
assert "yarn" in cmd
|
assert "yarn" in cmd
|
||||||
assert "cluster" in cmd
|
assert "cluster" in cmd
|
||||||
assert "--queue" in cmd and "default" 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"
|
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):
|
def test_run_spark_submit_returns_completed_process(monkeypatch):
|
||||||
fake = MagicMock()
|
fake = MagicMock()
|
||||||
fake.returncode = 0
|
fake.returncode = 0
|
||||||
|
|||||||
Reference in New Issue
Block a user