19. Spark Submit in PySpark
Dfspark-submit
spark-submit is the command-line tool for launching Spark applications. It handles dependency packaging, JVM configuration, and submission to cluster managers (YARN, Mesos, Kubernetes, Standalone).
DfDeployment Mode
A deployment mode determines where the driver runs: client mode (driver on the submitting machine, good for interactive/debugging) or cluster mode (driver on a cluster node, good for production).
Driver Memory Formula
Here,
- =Required driver memory
- =Memory needed to collect results from executors
- =Memory needed to store broadcast variables
- =Minimum recommended driver memory
Use cluster mode for production deployments — it ensures the driver survives if the submitting machine goes down. Use client mode for development and debugging where you need direct access to driver logs and the Spark UI.
Always set --conf spark.dynamicAllocation.enabled=true and --conf spark.shuffle.service.enabled=true for YARN cluster mode. This allows executors to scale up/down based on workload.
ThResource Contention
Theorem: The total resources requested by a Spark application must not exceed available_cluster_resources × F_{utilization} where F_{utilization} is the target utilization factor (typically 0.8–0.9 for shared clusters). Exceeding this causes resource contention and queueing delays.
- spark-submit handles packaging, JVM config, and cluster submission
- Client mode: driver on submit machine (debugging); Cluster mode: driver on cluster (production)
- Total resources = executors × (cores + memory + overhead)
- Set driver memory based on collect size and broadcast variable size
🏗️ Deployment Modes Architecture
Architecture Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ SPARK-SUBMIT DEPLOYMENT MODES │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CLIENT MODE │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Client Machine │───▶│ Cluster │ │ │
│ │ │ (spark-submit) │ │ Manager │ │ │
│ │ │ │ │ │ │ │
│ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │
│ │ │ │ Driver │ │ │ │ Executor │ │ │ │
│ │ │ │ (Local) │ │ │ │ 1 │ │ │ │
│ │ │ └───────────┘ │ │ └───────────┘ │ │ │
│ │ │ │ │ ┌───────────┐ │ │ │
│ │ │ Advantages: │ │ │ Executor │ │ │ │
│ │ │ • Easy debug │ │ │ 2 │ │ │ │
│ │ │ • Direct access│ │ └───────────┘ │ │ │
│ │ │ • Simple setup │ │ │ │ │
│ │ │ │ │ Disadvantages: │ │ │
│ │ │ Disadvantages: │ │ • Driver not │ │ │
│ │ │ • Single point │ │ protected │ │ │
│ │ │ of failure │ │ • Network │ │ │
│ │ │ • Network │ │ dependency │ │ │
│ │ │ dependency │ │ │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CLUSTER MODE │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Client Machine │───▶│ Cluster │ │ │
│ │ │ (spark-submit) │ │ Manager │ │ │
│ │ │ │ │ │ │ │
│ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │
│ │ │ │ (No │ │ │ │ Driver │ │ │ │
│ │ │ │ Driver) │ │ │ │ (Cluster) │ │ │ │
│ │ │ └───────────┘ │ │ └───────────┘ │ │ │
│ │ │ │ │ ┌───────────┐ │ │ │
│ │ │ Advantages: │ │ │ Executor │ │ │ │
│ │ │ • Fault │ │ │ 1 │ │ │ │
│ │ │ tolerant │ │ └───────────┘ │ │ │
│ │ │ • Better │ │ ┌───────────┐ │ │ │
│ │ │ resource │ │ │ Executor │ │ │ │
│ │ │ usage │ │ │ 2 │ │ │ │
│ │ │ │ │ └───────────┘ │ │ │
│ │ │ Disadvantages: │ │ │ │ │
│ │ │ • Harder to │ │ Advantages: │ │ │
│ │ │ debug │ │ • Driver │ │ │
│ │ │ • Complex setup│ │ protected │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ SPARK- SUBMIT FLOW │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 1. Submit Application │ │ │
│ │ │ spark-submit --master yarn --deploy-mode cluster │ │ │
│ │ │ app.py │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 2. Cluster Manager Allocates Resources │ │ │
│ │ │ • Requests driver container │ │ │
│ │ │ • Requests executor containers │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 3. Driver Starts and Manages Executors │ │ │
│ │ │ • Launches executor tasks │ │ │
│ │ │ • Monitors progress │ │ │
│ │ │ • Handles failures │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
📊 Configuration Parameters
Architecture Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ SPARK-SUBMIT CONFIGURATION PARAMETERS │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CORE PARAMETERS │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ --master Cluster master URL │ │ │
│ │ │ • local[*] │ │ │
│ │ │ • yarn │ │ │
│ │ │ • spark://host:port │ │ │
│ │ │ • k8s://https://host:port │ │ │
│ │ │ │ │ │
│ │ │ --deploy-mode Deployment mode │ │ │
│ │ │ • client (default) │ │ │
│ │ │ • cluster │ │ │
│ │ │ │ │ │
│ │ │ --name Application name │ │ │
│ │ │ --class Main class (for Java/Scala) │ │ │
│ │ │ --jars JAR files to include │ │ │
│ │ │ --py-files Python files to include │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ RESOURCE PARAMETERS │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ --driver-memory Driver memory (default: 1g) │ │ │
│ │ │ --driver-cores Driver cores (default: 1) │ │ │
│ │ │ --executor-memory Executor memory (default: 1g) │ │ │
│ │ │ --executor-cores Executor cores (default: 1) │ │ │
│ │ │ --num-executors Number of executors (YARN) │ │ │
│ │ │ --total-executors Total executors (YARN) │ │ │
│ │ │ │ │ │
│ │ │ Dynamic Allocation: │ │ │
│ │ │ --conf spark.dynamicAllocation.enabled=true │ │ │
│ │ │ --conf spark.dynamicAllocation.minExecutors=2 │ │ │
│ │ │ --conf spark.dynamicAllocation.maxExecutors=100 │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ RUNTIME PARAMETERS │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ --conf KEY=VALUE Set arbitrary Spark config │ │ │
│ │ │ --properties-file File with extra properties │ │ │
│ │ │ --driver-java-options Driver JVM options │ │ │
│ │ │ --executor-java-options Executor JVM options │ │ │
│ │ │ --driver-class-path Driver classpath │ │ │
│ │ │ --executor-class-path Executor classpath │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ YARN-SPECIFIC PARAMETERS │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ --queue YARN queue name │ │ │
│ │ │ --archives Archives to distribute │ │ │
│ │ │ --files Files to distribute │ │ │
│ │ │ --conf spark.yarn.maxAppAttempts=2 │ │ │
│ │ │ --conf spark.yarn.submit.waitAppCompletion=false │ │ │
│ │ │ --conf spark.yarn.historyServer.address=host:port │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
🔄 Execution Flow Diagram
Architecture Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ SPARK-SUBMIT EXECUTION FLOW │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 1: PARSE ARGUMENTS │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ spark-submit \ │ │ │
│ │ │ --master yarn \ │ │ │
│ │ │ --deploy-mode cluster \ │ │ │
│ │ │ --executor-memory 8g \ │ │ │
│ │ │ --executor-cores 4 \ │ │ │
│ │ │ --num-executors 10 \ │ │ │
│ │ │ --conf spark.sql.shuffle.partitions=200 \ │ │ │
│ │ │ my_app.py │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 2: VALIDATE CONFIGURATION │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Check master URL accessibility │ │ │
│ │ │ • Validate resource requests │ │ │
│ │ │ • Verify application dependencies │ │ │
│ │ │ • Check queue permissions (YARN) │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 3: SUBMIT TO CLUSTER MANAGER │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ YARN: │ │ │
│ │ │ • Create application master container │ │ │
│ │ │ • Upload application resources │ │ │
│ │ │ • Start application master │ │ │
│ │ │ │ │ │
│ │ │ Kubernetes: │ │ │
│ │ │ • Create driver pod │ │ │
│ │ │ • Pull container image │ │ │
│ │ │ • Start driver pod │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 4: DRIVER STARTUP │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Initialize SparkContext │ │ │
│ │ │ • Connect to cluster manager │ │ │
│ │ │ • Request executor resources │ │ │
│ │ │ • Set up communication channels │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 5: EXECUTE APPLICATION │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Distribute tasks to executors │ │ │
│ │ │ • Monitor task execution │ │ │
│ │ │ • Handle failures and retries │ │ │
│ │ │ • Collect results │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Step 6: CLEANUP AND EXIT │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Release executor resources │ │ │
│ │ │ • Clean up temporary files │ │ │
│ │ │ • Report final status │ │ │
│ │ │ • Exit with appropriate code │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
📚 Detailed Explanation
Spark-submit is the primary tool for deploying PySpark applications to cluster environments. It handles the complexity of resource allocation, dependency management, and application lifecycle, allowing developers to focus on business logic rather than cluster infrastructure. Understanding spark-submit's parameters and execution flow is essential for effective PySpark deployment.
The choice of deployment mode (client vs. cluster) significantly impacts application behavior. Client mode runs the driver on the submission machine, providing easy access to logs and debugging capabilities but creating a single point of failure. Cluster mode runs the driver within the cluster, offering better fault tolerance and resource efficiency but making debugging more challenging.
Resource configuration is critical for application performance and stability. Parameters like executor-memory, executor-cores, and num-executors determine how much computing power the application can utilize. Dynamic allocation allows the application to scale resources based on workload demands, improving cluster utilization for variable workloads.
Configuration management in spark-submit involves multiple layers: command-line arguments, configuration files, and programmatic settings. Understanding the precedence rules and configuration options helps optimize application behavior for different environments and workloads.
Dependency management is handled through parameters like --jars, --py-files, and --packages. These parameters ensure that all required libraries are available to the application, whether they are JAR files, Python modules, or external packages from repositories like Maven or PyPI.
YARN-specific parameters control integration with the Hadoop ecosystem, including queue management, application history tracking, and resource negotiation. Understanding these parameters is essential for running PySpark applications in production Hadoop environments.
Kubernetes deployment introduces cloud-native concepts like container orchestration, resource requests and limits, and pod management. Spark-submit's Kubernetes integration enables running PySpark applications in modern containerized environments with auto-scaling and resource isolation capabilities.
Best practices for spark-submit include: starting with conservative resource configurations and tuning based on workload characteristics, using dynamic allocation for variable workloads, implementing proper dependency management, testing configurations in staging environments, and monitoring application performance in production.
Advanced techniques include custom resource schedulers, application templates for common configurations, and automated deployment pipelines. These techniques help organizations standardize and optimize their PySpark deployment processes.
📊 Key Concepts Table
| Parameter | Description | Default | Recommended |
|---|---|---|---|
| --master | Cluster master URL | local[*] | yarn, k8s://... |
| --deploy-mode | Driver deployment mode | client | cluster (production) |
| --executor-memory | Memory per executor | 1g | 4-16g (workload dependent) |
| --executor-cores | Cores per executor | 1 | 4-8 (workload dependent) |
| --num-executors | Number of executors | 2 | 10-100 (workload dependent) |
| --conf | Custom configuration | - | As needed |
💻 Code Examples
Basic spark-submit Command
# Basic spark-submit with Python application
spark-submit \
--master yarn \
--deploy-mode client \
--name "MySparkApplication" \
--executor-memory 8g \
--executor-cores 4 \
--num-executors 10 \
--conf spark.sql.shuffle.partitions=200 \
--conf spark.default.parallelism=200 \
my_application.py
# spark-submit with dependencies
spark-submit \
--master yarn \
--deploy-mode cluster \
--name "ApplicationWithDeps" \
--py-files utilities.py,helpers.zip \
--jars mysql-connector-java-8.0.27.jar \
--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 \
--executor-memory 16g \
--executor-cores 8 \
--driver-memory 8g \
--conf spark.dynamicAllocation.enabled=true \
--conf spark.dynamicAllocation.maxExecutors=50 \
my_application.py
# spark-submit with YARN queue
spark-submit \
--master yarn \
--deploy-mode cluster \
--queue production \
--name "ProductionJob" \
--executor-memory 8g \
--executor-cores 4 \
--num-executors 20 \
--conf spark.yarn.maxAppAttempts=3 \
--conf spark.yarn.submit.waitAppCompletion=false \
production_job.py
Kubernetes Deployment
# Kubernetes deployment with spark-submit
spark-submit \
--master k8s://https://kubernetes.default.svc:6443 \
--deploy-mode cluster \
--name "KubernetesSparkApp" \
--conf spark.kubernetes.container.image=spark:3.3.0 \
--conf spark.kubernetes.namespace=spark-jobs \
--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
--conf spark.kubernetes.driver.label.app=spark-processing \
--conf spark.kubernetes.executor.label.app=spark-processing \
--conf spark.kubernetes.executor.request.cores=2 \
--conf spark.kubernetes.executor.limit.cores=4 \
--conf spark.kubernetes.driver.request.cores=1 \
--conf spark.kubernetes.driver.limit.cores=2 \
--conf spark.executor.memory=8g \
--conf spark.driver.memory=4g \
local:///opt/spark/examples/jars/spark-examples_2.12-3.3.0.jar
# Kubernetes with dynamic allocation
spark-submit \
--master k8s://https://kubernetes.default.svc:6443 \
--deploy-mode cluster \
--name "KubernetesDynamic" \
--conf spark.kubernetes.container.image=spark:3.3.0 \
--conf spark.kubernetes.namespace=spark-jobs \
--conf spark.dynamicAllocation.enabled=true \
--conf spark.dynamicAllocation.minExecutors=2 \
--conf spark.dynamicAllocation.maxExecutors=20 \
--conf spark.dynamicAllocation.executorIdleTimeout=60s \
--conf spark.kubernetes.allocation.batch.size=5 \
local:///opt/spark/examples/jars/spark-examples_2.12-3.3.0.jar
Python Application with Configuration
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import sys
import os
def create_spark_session(app_name, configs=None):
"""Create Spark session with configuration"""
builder = SparkSession.builder.appName(app_name)
# Apply custom configurations
if configs:
for key, value in configs.items():
builder = builder.config(key, value)
# Set default configurations
builder = builder \
.config("spark.sql.shuffle.partitions", "200") \
.config("spark.sql.autoBroadcastJoinThreshold", "10485760") \
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
.config("spark.kryoserializer.buffer.max", "512m")
return builder.getOrCreate()
def main():
"""Main application entry point"""
# Parse command line arguments
if len(sys.argv) < 2:
print("Usage: spark-submit my_app.py <input_path> [output_path]")
sys.exit(1)
input_path = sys.argv[1]
output_path = sys.argv[2] if len(sys.argv) > 2 else "/tmp/output"
# Create Spark session
spark = create_spark_session(
app_name="PySparkApplication",
configs={
"spark.executor.memory": "8g",
"spark.executor.cores": "4",
"spark.dynamicAllocation.enabled": "true"
}
)
try:
# Read data
df = spark.read.parquet(input_path)
# Process data
result = df \
.withColumn("processed_at", current_timestamp()) \
.filter(col("value") > 0) \
.groupBy("category") \
.agg(
sum("value").alias("total_value"),
count("*").alias("count"),
avg("value").alias("avg_value")
)
# Write results
result.write \
.mode("overwrite") \
.parquet(output_path)
print(f"Application completed successfully. Output: {output_path}")
except Exception as e:
print(f"Application failed: {str(e)}")
sys.exit(1)
finally:
spark.stop()
if __name__ == "__main__":
main()
Configuration File Example
# spark-submit.properties
# Cluster Configuration
spark.master=yarn
spark.submit.deployMode=cluster
spark.app.name=ProductionSparkApp
# Resource Configuration
spark.executor.memory=16g
spark.executor.cores=8
spark.driver.memory=8g
spark.driver.cores=4
# Dynamic Allocation
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=5
spark.dynamicAllocation.maxExecutors=100
spark.dynamicAllocation.executorIdleTimeout=60s
spark.shuffle.service.enabled=true
# Performance Configuration
spark.sql.shuffle.partitions=400
spark.sql.autoBroadcastJoinThreshold=20971520
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max=1024m
# YARN Configuration
spark.yarn.queue=production
spark.yarn.maxAppAttempts=3
spark.yarn.submit.waitAppCompletion=false
spark.yarn.historyServer.address=history-server:18080
# Monitoring Configuration
spark.eventLog.enabled=true
spark.eventLog.dir=hdfs:///spark-history
spark.history.fs.logDirectory=hdfs:///spark-history
spark.history.ui.port=18080
📈 Performance Metrics
| Metric | Target | Warning | Critical | Optimization |
|---|---|---|---|---|
| Startup Time | < 30s | 30-60s | > 60s | Reduce dependencies, optimize configuration |
| Resource Utilization | > 80% | 60-80% | < 60% | Tune parallelism, use dynamic allocation |
| Application Success Rate | > 99% | 95-99% | < 95% | Improve error handling, increase retries |
| Task Completion Time | < 5min | 5-15min | > 15min | Optimize code, increase resources |
| Cluster Utilization | > 70% | 50-70% | < 50% | Use dynamic allocation, optimize scheduling |
🏆 Best Practices
- Use cluster mode for production - Better fault tolerance and resource management
- Configure dynamic allocation - Optimize resource usage for variable workloads
- Test configurations in staging - Validate settings before production deployment
- Monitor application performance - Track key metrics and optimize as needed
- Implement proper dependency management - Ensure all dependencies are available
- Use appropriate resource configurations - Balance performance with cost
- Configure logging and monitoring - Enable event logs and history server
- Handle failures gracefully - Implement retry mechanisms and error handling
- Document configurations - Maintain clear documentation of settings
- Review and optimize regularly - Continuously improve based on performance data
🔗 Related Topics
- 17-cluster-management.mdx: Cluster resource management and optimization
- 18-gc-tuning.mdx: Garbage collection and memory optimization
- 20-monitoring-metrics.mdx: Monitoring application performance
- 11-structured-streaming.mdx: Streaming application deployment
See Also
- Kafka Streams (kafka/03): Deployment patterns for Kafka stream processing
- Data Engineering Streaming (data-engineering/022): Production deployment of streaming pipelines