What is the difference between --jars, --packages, and --py-files?

--jars takes local jar paths and ships them to executors. --packages takes Maven coordinates and resolves them, including transitive dependencies, at submit time. --py-files ships Python modules as a zip or egg so imports resolve on executors.

Why does my job work in spark-shell but fail on submit?

Usually a dependency or a variable that exists in your shell and not in the cluster. The shell has your local classpath, environment variables, and files; a submitted job only has what you shipped. Check for hardcoded local paths and unpackaged imports first.

How do I choose executor count and size?

Start from the cluster's node shape. Leave one core and about 1 GB per node for the OS and node manager, then divide the rest into executors of four to five cores. Very large executors suffer long garbage collection pauses; very small ones waste JVM overhead.

Should I use dynamic allocation?

Usually yes for shared clusters. Spark requests executors as stages need them and releases idle ones, so a job with an uneven shape does not hold resources it is not using. It requires an external shuffle service so shuffle files survive executor removal.

Deploying Spark Jobs with spark-submit

Everything works in spark-shell. Then you submit the same logic to a cluster and it fails in ways the shell never showed you. This lesson is about that gap.

A submittable application

# app.py
import sys
from pyspark.sql import SparkSession, functions as F


def main(input_path: str, output_path: str) -> None:
    spark = SparkSession.builder.appName("daily-revenue").getOrCreate()

    # No .master() call — the cluster supplies it via spark-submit.
    orders = spark.read.parquet(input_path)

    result = (
        orders.filter(F.col("amount") > 0)
              .groupBy("country")
              .agg(F.round(F.sum("amount"), 2).alias("revenue"),
                   F.count("*").alias("orders"))
              .orderBy(F.desc("revenue"))
    )

    result.show(truncate=False)
    result.coalesce(1).write.mode("overwrite").parquet(output_path)

    print(f"wrote {result.count()} rows to {output_path}")
    spark.stop()


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("usage: app.py <input> <output>", file=sys.stderr)
        sys.exit(1)
    main(sys.argv[1], sys.argv[2])

Note what is absent: no .master("local[*]"). Hardcoding the master means the job ignores whatever the cluster tells it and is the most common reason a submitted job silently runs on one machine.

Submitting locally first

spark-submit --master "local[4]" app.py data/orders_parquet /tmp/out

25/02/15 09:41:02 INFO SparkContext: Running Spark version 3.5.4
25/02/15 09:41:02 INFO ResourceUtils: Resources for spark.driver: 
25/02/15 09:41:03 INFO SparkContext: Submitted application: daily-revenue
+-------+-------+------+
|country|revenue|orders|
+-------+-------+------+
|US     |590.7  |2     |
|UK     |570.75 |2     |
|DE     |95.0   |1     |
+-------+-------+------+
wrote 3 rows to /tmp/out
25/02/15 09:41:09 INFO SparkContext: Successfully stopped SparkContext

Always do this before touching a cluster — it catches logic errors without waiting for resource allocation.

Submitting to YARN

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name daily-revenue \
  --driver-memory 4g \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.dynamicAllocation.enabled=false \
  app.py s3a://data-lake/orders/ s3a://data-lake/reports/revenue/

25/02/15 09:52:11 INFO Client: Requesting a new application from cluster with 24 NodeManagers
25/02/15 09:52:11 INFO Client: Verifying our application has not requested more than the maximum memory capability of the cluster (24576 MB per container)
25/02/15 09:52:11 INFO Client: Will allocate AM container, with 4505 MB memory including 409 MB overhead
25/02/15 09:52:12 INFO Client: Submitting application application_1771000000_0087 to ResourceManager
25/02/15 09:52:13 INFO Client: Application report for application_1771000000_0087 (state: ACCEPTED)
25/02/15 09:52:24 INFO Client: Application report for application_1771000000_0087 (state: RUNNING)
25/02/15 09:54:41 INFO Client: Application report for application_1771000000_0087 (state: FINISHED)
	 final status: SUCCEEDED
	 tracking URL: http://rm.internal:8088/proxy/application_1771000000_0087/

Notice 4505 MB memory including 409 MB overhead — you asked for 4g and YARN allocated more. That overhead is not optional, and forgetting it is why capacity planning based on --executor-memory alone comes up short.

In cluster mode the show() output never reaches your terminal. Fetch it:

yarn logs -applicationId application_1771000000_0087 | grep -A6 country

+-------+----------+--------+
|country|revenue   |orders  |
+-------+----------+--------+
|US     |8421904.55|  120483|
|UK     |7190228.10|  104217|
|DE     |3388417.92|   51902|
+-------+----------+--------+

Failure 1: the dependency that only exists locally

spark-submit --master yarn --deploy-mode cluster app_with_deps.py ...

25/02/15 10:02:44 ERROR Executor: Exception in task 3.0 in stage 2.0 (TID 47)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/mnt/yarn/usercache/.../pyspark/worker.py", line 830, in main
    process()
  File "app_with_deps.py", line 12, in enrich
    import ujson
ModuleNotFoundError: No module named 'ujson'

Installed on your machine, not on the executors. Ship it:

# Option A: a zip of pure-Python modules
zip -r deps.zip mypackage/
spark-submit --py-files deps.zip app.py ...

# Option B: a packed conda/venv environment, for compiled dependencies
venv-pack -o pyspark_env.tar.gz
spark-submit \
  --archives pyspark_env.tar.gz#environment \
  --conf spark.pyspark.python=./environment/bin/python \
  app.py ...

25/02/15 10:11:03 INFO Client: Uploading resource file:/home/you/pyspark_env.tar.gz -> hdfs://.../pyspark_env.tar.gz
25/02/15 10:13:52 INFO Client: Application report ... final status: SUCCEEDED

--py-files only works for pure Python. Anything with a compiled extension — NumPy, pandas, ujson — needs the packed-environment approach.

Failure 2: the missing connector jar

py4j.protocol.Py4JJavaError: An error occurred while calling o42.load.
: java.lang.ClassNotFoundException: Failed to find data source: kafka.
Please find packages at https://spark.apache.org/third-party-projects.html

Spark does not bundle connectors. Add them by Maven coordinate:

spark-submit \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.4,org.apache.hadoop:hadoop-aws:3.3.4 \
  app.py ...

:: resolution report :: resolve 1842ms :: artifacts dl 63ms
	:: modules in use:
	org.apache.hadoop#hadoop-aws;3.3.4 from central in [default]
	org.apache.spark#spark-sql-kafka-0-10_2.12;3.5.4 from central in [default]
	---------------------------------------------------------------------
	|                  |            modules            ||   artifacts   |
	|       conf       | number| search|dwnlded|evicted|| number|dwnlded|
	---------------------------------------------------------------------
	|      default     |   14  |   0   |   0   |   0   ||   14  |   0   |
	---------------------------------------------------------------------

The Scala version in the artifact name — _2.12 — must match your Spark build. A _2.13 artifact on a 2.12 cluster produces a NoSuchMethodError at runtime rather than a clean resolution failure, which is much harder to diagnose.

Failure 3: the container killed for memory

25/02/15 10:31:17 ERROR YarnScheduler: Lost executor 7 on worker-12.internal:
Container killed by YARN for exceeding physical memory limits.
9.4 GB of 9.0 GB physical memory used. Consider boosting spark.executor.memoryOverhead.

The JVM heap was within --executor-memory; the container was not. Overhead covers off-heap buffers, the shuffle service, and — in PySpark — the Python worker processes, which live entirely outside the JVM heap.

spark-submit \
  --executor-memory 8g \
  --conf spark.executor.memoryOverhead=2g \
  --conf spark.executor.pyspark.memory=2g \
  app.py ...

For PySpark jobs doing heavy pandas work, overhead often needs to be 25-40% of executor memory rather than the 10% default.

Kubernetes

spark-submit \
  --master k8s://https://k8s-api.internal:6443 \
  --deploy-mode cluster \
  --name daily-revenue \
  --conf spark.kubernetes.container.image=registry.internal/spark-app:3.5.4 \
  --conf spark.kubernetes.namespace=analytics \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  --conf spark.executor.instances=10 \
  local:///opt/app/app.py s3a://data-lake/orders/ s3a://data-lake/reports/

25/02/15 10:45:02 INFO LoggingPodStatusWatcherImpl: State changed, new state:
	 pod name: daily-revenue-8f2a1c94-driver
	 namespace: analytics
	 phase: Pending
25/02/15 10:45:14 INFO LoggingPodStatusWatcherImpl: State changed, new state:
	 phase: Running
25/02/15 10:47:38 INFO LoggingPodStatusWatcherImpl: Container final statuses:
	 exit code: 0
	 phase: Succeeded

local:// means “a path inside the image”, not your machine — the application must already be baked into the container.

Dynamic allocation

A job whose stages need different amounts of parallelism wastes resources with a fixed executor count:

spark-submit \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=2 \
  --conf spark.dynamicAllocation.maxExecutors=50 \
  --conf spark.dynamicAllocation.executorIdleTimeout=60s \
  --conf spark.shuffle.service.enabled=true \
  app.py ...

25/02/15 11:02:41 INFO ExecutorAllocationManager: Requesting 12 new executors because tasks are backlogged
25/02/15 11:04:18 INFO ExecutorAllocationManager: Request to remove executorIds: 14, 15, 16 (idle for 60s)

Spark scaled to 14 executors for the heavy stage and released them when the tail ran on two. The external shuffle service is required — without it, removing an executor destroys the shuffle files it was serving.

Sizing from node shape

For nodes with 16 cores and 64 GB:

reserve for OS + node manager:   1 core,  1 GB
available per node:             15 cores, 63 GB

executors per node at 5 cores:   3
memory per executor:            63 / 3 = 21 GB
minus 10% overhead:             ~18 GB heap

--executor-cores 5 --executor-memory 18g
--num-executors (3 × node_count)

Five cores is the conventional ceiling — beyond it, HDFS and object-store clients contend and throughput per core drops.

Practice

1. Hardcode .master("local[*]") and submit to a cluster.

INFO SparkContext: Running Spark version 3.5.4
INFO Executor: Starting executor ID driver on host worker-03.internal

It runs — on one node, using one machine’s cores, while the cluster sits idle. No error, just a job that is inexplicably slow. Never set the master in application code.

2. Submit without a required jar, then with --packages.

without: java.lang.ClassNotFoundException: Failed to find data source: kafka
with:    :: modules in use: org.apache.spark#spark-sql-kafka-0-10_2.12;3.5.4

--packages resolves transitive dependencies too, which is why it is preferable to hunting down individual jars for --jars.

3. Set --executor-memory 8g and check the container size YARN allocates.

Will allocate AM container, with 9011 MB memory including 819 MB overhead

About 10% more than requested. Plan cluster capacity on container size, not heap size, or you will fit fewer executors per node than you expected.

4. Enable dynamic allocation without the external shuffle service.

org.apache.spark.SparkException: Dynamic allocation of executors requires the external
shuffle service. You may enable this through spark.shuffle.service.enabled.

Spark refuses at startup rather than losing shuffle data later. On Kubernetes there is an alternative — spark.dynamicAllocation.shuffleTracking.enabled keeps executors alive while their shuffle output is still needed.

Next: reading the Spark UI to find out where the time actually went.