Apache Spark User Guide

Apache Spark is a unified analytics engine for large-scale data processing. It provides high-level APIs in Scala, Java, Python, and R (R support is deprecated), backed by an optimized execution engine that supports general computation graphs. On top of the core engine, Spark ships a rich set of higher-level libraries: Spark SQL for structured data and DataFrames, the pandas API on Spark for pandas workloads, MLlib for machine learning, GraphX for graph processing, and Structured Streaming for stream processing.

This guide is the end-user reference. It covers installing Spark, launching the interactive shells, submitting applications with spark-submit, working with the core data APIs, and using the higher-level libraries. Operational topics — cluster deployment, security hardening, memory tuning, and internals — live in the companion Admin Guide.

Note

This documentation is generated from the development (master) branch, which self-reports version 5.0.0-SNAPSHOT (PySpark 5.0.0.dev0). The branch builds against Scala 2.13.18 and requires Java 17 or later. Feature names and defaults reflect the master branch at the documented commit; released versions (4.x and earlier) may differ. Always cross-check the official configuration guide for your exact release.

1. Quick Start

The fastest way to try Spark is through an interactive shell. From an unpacked Spark distribution:

Scala

./bin/spark-shell

scala> spark.range(1000 * 1000 * 1000).count()
res0: Long = 1000000000

Python

./bin/pyspark

>>> spark.range(1000 * 1000 * 1000).count()
1000000000

SQL

./bin/spark-sql

spark-sql> SELECT count(*) FROM range(1000000000);
1000000000

Both shells start a local Spark session bound to a pre-created spark object (a SparkSession). The command above builds a distributed range of one billion rows and counts them — locally, across all your CPU cores.

Tip

Every interactive shell defaults to running locally with one thread per core (local[*]). You do not need a cluster to learn or develop with Spark. Point at a real cluster later with --master (see §5).

To run a bundled example:

./bin/run-example SparkPi

Pi is roughly 3.1417...

2. Requirements & Installation

2.1 System requirements

Important

The spark-submit and spark-class launchers locate the runtime via the SPARK_HOME environment variable (resolved automatically by the bundled find-spark-home script if unset) and a Java binary from JAVA_HOME (falling back to java on the PATH). If neither is available the launcher exits with JAVA_HOME is not set.

2.2 Install options

Pre-built distribution

Download a pre-built package from the Spark downloads page, unpack it, and run the bin/ scripts directly. No build step is required.

tar -xzf spark-<version>-bin-hadoop3.tgz
cd spark-<version>-bin-hadoop3
./bin/spark-shell

PySpark via pip

For Python users, install PySpark from PyPI:

pip install pyspark

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("demo").getOrCreate()
spark.range(5).show()

The pip package bundles its own Spark JARs, so no separate distribution is needed for local use.

Build from source

Spark builds with Apache Maven via the wrapper at build/mvn:

./build/mvn -DskipTests clean package

You do not need this step if you downloaded a pre-built package. See the official Building Spark guide for profiles (Hadoop version, YARN, Hive).

2.3 Anatomy of a distribution

2.4 A note about Hadoop versions

Spark uses the Hadoop client library to talk to HDFS and other Hadoop-supported storage. Because the protocols differ between Hadoop versions, build (or download) Spark against the same Hadoop version your cluster runs. Pre-built packages are published for common Hadoop lines; for custom distributions, build from source with the appropriate Hadoop profile.

Warning

Mixing a Spark build compiled for one Hadoop line against a cluster running a different line is a common source of cryptic NoSuchMethodError / ClassNotFoundException failures at runtime.

3. Core Concepts

You interact with Spark through a small number of foundational objects. This section is a working mental model; the Admin Guide covers the execution internals.

SparkSession : The unified entry point for all Spark functionality. In the shells it is pre-created as spark. In an application you build it yourself:

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("My Application")
         .master("local[*]")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

SparkContext : The lower-level handle to the cluster (available as spark.sparkContext, or sc in the shells). It is the entry point to the RDD API.

DataFrame / Dataset : A distributed table of rows with a named schema — the primary, optimized data abstraction. DataFrames are processed by the Catalyst optimizer and the Tungsten execution engine.

RDD (Resilient Distributed Dataset) : The original low-level abstraction: an immutable, partitioned collection. Prefer DataFrames/Datasets unless you need fine-grained control.

Note

Transformations are lazy. Operations like select, filter, and join build a logical plan but do not execute. Work runs only when you call an action such as count(), collect(), show(), or write. This lets Spark optimize the whole pipeline before executing it.

4. The Interactive Shells

Spark ships several REPLs in bin/. All of them ultimately launch through spark-class and accept the same spark-submit options described in §5 (for example --master, --conf, --jars, --packages).

4.1 spark-shell (Scala)

./bin/spark-shell --master "local[4]"

scala> val df = spark.read.json("examples/src/main/resources/people.json")
scala> df.filter($"age" > 20).show()

4.2 pyspark (Python)

./bin/pyspark --master "local[4]"

>>> df = spark.read.json("examples/src/main/resources/people.json")
>>> df.filter(df.age > 20).show()

Tip

Behind the scenes, pyspark honors several environment variables: PYSPARK_PYTHON (the Python executable to use, default python3), PYSPARK_DRIVER_PYTHON, and PYSPARK_SUBMIT_ARGS (default pyspark-shell). Set PYSPARK_PYTHON to pin a specific interpreter across driver and executors.

4.3 spark-sql

./bin/spark-sql

spark-sql> CREATE TABLE people (name STRING, age INT) USING parquet;
spark-sql> INSERT INTO people VALUES ('Alice', 30), ('Bob', 25);
spark-sql> SELECT name FROM people WHERE age > 26;
Alice

spark-sql accepts the standard Spark CLI options plus its own SQL-CLI options (Spark prints these when you pass --help to a SQL shell launched through spark-submit).

4.4 Connecting a remote session (Spark Connect)

Spark Connect lets a thin client drive a remote Spark cluster over gRPC. Point any shell at a Connect endpoint with --remote:

./bin/pyspark --remote "sc://localhost:15002"

./bin/spark-connect-shell

Important

--remote is mutually exclusive with --master and --deploy-mode. Setting more than one produces the error Remote cannot be specified with master and/or deploy mode. See §9 for details.

5. Submitting Applications with spark-submit

spark-submit is the universal launcher for Spark applications. It supports every cluster manager (local, Standalone, YARN, Kubernetes) and every language binding through one consistent interface.

5.1 Usage

Usage: spark-submit [options] <app jar | python file | R file> [app arguments]
Usage: spark-submit --kill [submission ID] --master [spark://...]
Usage: spark-submit --status [submission ID] --master [spark://...]
Usage: spark-submit run-example [options] example-class [example args]

Examples:

Python application

./bin/spark-submit \
  --master "local[8]" \
  --name "wordcount" \
  my_app.py input.txt

Scala/Java JAR

./bin/spark-submit \
  --class com.example.MyApp \
  --master spark://master:7077 \
  --deploy-mode cluster \
  --executor-memory 4G \
  --total-executor-cores 16 \
  myapp-assembly.jar arg1 arg2

On YARN

./bin/spark-submit \
  --class com.example.MyApp \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 8G \
  --queue production \
  myapp.jar

On Kubernetes

./bin/spark-submit \
  --master k8s://https://k8s-apiserver:6443 \
  --deploy-mode cluster \
  --name spark-job \
  --conf spark.kubernetes.container.image=myrepo/spark:latest \
  --conf spark.executor.instances=5 \
  local:///opt/app/myapp.jar

Important

You must always specify a primary resource — a JAR, a Python file, or an R file. Omitting it fails with Must specify a primary resource (JAR or Python or R file).

5.2 General options

5.3 Resource options

5.4 Cluster-manager-specific options

Note

When --master starts with yarn, either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment, or submission fails with When running with master 'yarn' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.

5.5 Master URL formats

6. Working with Data: DataFrames & Spark SQL

Spark SQL is the workhorse for structured data. You can mix the DataFrame API and raw SQL freely against the same engine.

6.1 Reading and writing data

# Read
df = spark.read.parquet("/data/events")
csv = spark.read.option("header", "true").csv("/data/users.csv")
js  = spark.read.json("/data/logs.json")

# Write
df.write.mode("overwrite").parquet("/out/events")
df.write.format("json").save("/out/events_json")

The built-in data sources include Parquet (the default), ORC, JSON, CSV, text, Avro and Protobuf (via the bundled connectors), and JDBC. The default source is controlled by spark.sql.sources.default (parquet).

6.2 Transformations

from pyspark.sql import functions as F

result = (df
          .filter(F.col("country") == "US")
          .groupBy("category")
          .agg(F.sum("amount").alias("total"),
               F.countDistinct("user_id").alias("users"))
          .orderBy(F.desc("total")))

result.show()

6.3 Running SQL

Register a DataFrame as a temporary view, then query it:

df.createOrReplaceTempView("events")
spark.sql("""
    SELECT category, SUM(amount) AS total
    FROM events
    WHERE country = 'US'
    GROUP BY category
    ORDER BY total DESC
""").show()

6.4 Tables and the catalog

CREATE TABLE sales (id BIGINT, amount DOUBLE) USING parquet;
INSERT INTO sales VALUES (1, 9.99), (2, 19.50);
SELECT * FROM sales;

Managed tables are stored under the warehouse directory (spark.sql.warehouse.dir). The catalog backend is selected by spark.sql.catalogImplementation (in-memory or hive).

Tip

The single most common performance lever for SQL/DataFrame jobs is spark.sql.shuffle.partitions (the number of partitions after a shuffle). Combine it with Adaptive Query Execution (spark.sql.adaptive.enabled), which coalesces shuffle partitions and handles skewed joins automatically at runtime. See the Admin Guide for the full tuning reference.

6.5 Arrow-accelerated Python ↔ Spark conversion

When converting between PySpark and pandas, enable Arrow for a large speedup:

spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
pdf = df.toPandas()                # fast, columnar transfer
sdf = spark.createDataFrame(pdf)   # fast, columnar transfer

7. Higher-Level Libraries

7.1 Structured Streaming

Structured Streaming treats a live data stream as an unbounded table you query with the same DataFrame API:

lines = (spark.readStream
         .format("socket")
         .option("host", "localhost")
         .option("port", 9999)
         .load())

counts = lines.groupBy("value").count()

query = (counts.writeStream
         .outputMode("complete")
         .format("console")
         .start())

query.awaitTermination()

Sources and sinks include sockets, files, rate generators, and Kafka (via the bundled kafka-0-10-sql connector) and Kinesis (via kinesis-asl).

7.2 pandas API on Spark

Run pandas code at Spark scale with a near drop-in API:

import pyspark.pandas as ps

psdf = ps.read_parquet("/data/events")
psdf["amount"].sum()
psdf.groupby("category")["amount"].mean()

This is ideal for migrating existing pandas workloads onto a cluster with minimal rewrites.

7.3 MLlib (machine learning)

The spark.ml DataFrame-based API provides feature transformers, estimators, and pipelines:

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression
from pyspark.ml import Pipeline

assembler = VectorAssembler(inputCols=["x1", "x2"], outputCol="features")
lr = LinearRegression(featuresCol="features", labelCol="y")
model = Pipeline(stages=[assembler, lr]).fit(train_df)
predictions = model.transform(test_df)

MLlib ships a large library of feature transformers (under org.apache.spark.ml.feature), classification/regression/clustering algorithms, and model evaluation utilities.

7.4 GraphX

GraphX is the Scala graph-processing library, including built-in algorithms such as PageRank, Connected Components, Triangle Count, SVD++, and Shortest Paths (in org.apache.spark.graphx.lib):

import org.apache.spark.graphx.GraphLoader

val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
val ranks  = graph.pageRank(0.0001).vertices
ranks.collect().foreach(println)

7.5 Running the bundled examples

./bin/run-example SparkPi 100
./bin/run-example sql.SparkSQLExample
./bin/run-example ml.LogisticRegressionExample

Set MASTER to submit an example to a cluster instead of running locally:

MASTER=spark://host:7077 ./bin/run-example SparkPi

Many example programs print usage help when run with no parameters.

8. Declarative Pipelines (spark-pipelines)

The master branch includes Spark Declarative Pipelines, driven by the bin/spark-pipelines launcher and the pyspark.pipelines module. You define datasets in a pipeline spec and let Spark resolve the dependency graph and materialize them.

./bin/spark-pipelines init --name my_pipeline
cd my_pipeline
./bin/spark-pipelines run --spec pipeline.yml

Note

Declarative Pipelines is a newer capability on the development branch. Confirm availability and the exact spec format against the release notes for your Spark version.

9. Spark Connect

Spark Connect decouples the client from the Spark driver: a lightweight client communicates with a remote Connect server over gRPC, so you can run Spark code from notebooks, IDEs, or thin applications without colocating with the cluster.

Start a Connect server (see Admin Guide for production setup):

./sbin/start-connect-server.sh

Connect a client:

PySpark

from pyspark.sql import SparkSession
spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
spark.range(10).show()

Shell

./bin/pyspark --remote "sc://localhost:15002"

You can also set the SPARK_REMOTE environment variable instead of passing --remote.

Important

--remote cannot be combined with --master or --deploy-mode. The two execution models are mutually exclusive.

10. Configuration Basics

Spark configuration properties (the spark.* keys) can be set three ways, in increasing precedence:

1. conf/spark-defaults.conf — site-wide defaults (copy from spark-defaults.conf.template).
2. --conf KEY=VALUE on the spark-submit/shell command line.
3. Programmatically via SparkSession.builder.config(...) or spark.conf.set(...) at runtime (for properties that are runtime-mutable).

Example spark-defaults.conf:

spark.master                     spark://master:7077
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://namenode:8021/directory
spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.driver.memory              5g
spark.executor.extraJavaOptions  -XX:+PrintGCDetails -Dkey=value

A few of the most commonly used properties:

Tip

The complete configuration reference — hundreds of spark.* properties grouped by subsystem (memory, shuffle, dynamic allocation, scheduling, security, networking, SQL, YARN, Kubernetes) — is in the Admin Guide. For end-user tuning, the SQL and memory sections cover 90% of cases.

11. Troubleshooting

The errors below are validated directly by Spark's argument parser and submission code.

Error: Must specify a primary resource (JAR or Python or R file)

Cause: spark-submit was invoked without an application file. Solution: Provide a JAR, .py, or .R file as the last positional argument (before app arguments). Prevention: Follow the form spark-submit [options] <app> [app args].

Error: When running with master 'yarn' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.

Cause: YARN submission with no Hadoop/YARN configuration directory exported. Solution: export HADOOP_CONF_DIR=/etc/hadoop/conf (or YARN_CONF_DIR) before submitting. Prevention: Set it in conf/spark-env.sh.

Error: Remote cannot be specified with master and/or deploy mode.

Cause: --remote was combined with --master and/or --deploy-mode. Solution: Use either --remote (Spark Connect) or --master/--deploy-mode (classic).

Error: Driver memory must be a positive number / Executor memory must be a positive number

Cause: A non-positive or unparseable memory value (e.g. --driver-memory 0, or a malformed size). Solution: Use a positive size with a unit, e.g. 2g, 1500m. Prevention: Sizes accept k/m/g/t suffixes; 2G means 2 gibibytes.

Error: Driver cores must be a positive number / Executor cores must be a positive number / Total executor cores must be a positive number

Cause: A non-positive core count. Solution: Pass a positive integer to --driver-cores, --executor-cores, or --total-executor-cores.

Error: Number of executors must be a positive number

Cause: --num-executors set to a non-positive value while dynamic allocation is disabled. Solution: Use a positive integer, or enable dynamic allocation (spark.dynamicAllocation.enabled=true).

Error: Only one of --proxy-user or --principal can be provided.

Cause: Both impersonation (--proxy-user) and Kerberos login (--principal) were given. Solution: Choose one authentication path.

Error: JAVA_HOME is not set (from spark-class)

Cause: No Java on JAVA_HOME or PATH. Solution: Install Java 17+ and export JAVA_HOME=/path/to/jdk.

Symptom: Out-of-memory / GC thrashing in the driver or executors

Cause: Heap too small for the workload, or too much data pulled to the driver (collect() on a huge DataFrame). Solution: Increase --driver-memory / --executor-memory; avoid collect() on large data (use show(), take(n), or write to storage). Cap driver results with spark.driver.maxResultSize.

Symptom: Job is slow with many small or skewed shuffle partitions

Cause: Default spark.sql.shuffle.partitions (200) doesn't fit your data size, or skewed keys. Solution: Keep Adaptive Query Execution on (spark.sql.adaptive.enabled=true); it coalesces partitions and splits skewed joins. Tune spark.sql.shuffle.partitions for very large jobs.

12. FAQ

Q: Do I need a cluster to use Spark? No. The shells and spark-submit default to local[*], running across your local cores. Add --master when you have a cluster.

Q: What's the difference between a DataFrame and an RDD? DataFrames are higher-level, schema-aware, and optimized by Catalyst/Tungsten. RDDs are lower-level and untyped at the row level. Prefer DataFrames unless you need fine-grained control.

Q: How do I add a third-party library to my job? Use --packages groupId:artifactId:version for Maven artifacts, or --jars for local JARs, or --py-files for Python dependencies.

Q: Which Python versions are supported? PySpark is tested against Python 3.11–3.14 (including the free-threaded nogil build) on the development branch. Pin the interpreter with PYSPARK_PYTHON.

Q: client vs cluster deploy mode? client runs the driver where you launched spark-submit (good for interactive use). cluster runs the driver on a cluster node (good for production submission, decouples from your terminal).

Q: How do I see what my job did after it finished? Enable event logging (spark.eventLog.enabled=true, spark.eventLog.dir=...) and browse the History Server (./sbin/start-history-server.sh). See the Admin Guide.

Q: Is SparkR still supported? R support is deprecated. New work should use PySpark, Scala, or SQL.

Q: What is Spark Connect and when should I use it? A gRPC client/server split that lets thin clients (notebooks, IDEs) drive a remote cluster. Use it when you don't want to colocate your application with the driver.

Q: How do I run SQL without writing any code? Use ./bin/spark-sql for an interactive SQL prompt, or start the Thrift JDBC/ODBC server (./sbin/start-thriftserver.sh) and connect with ./bin/beeline or any JDBC client.

Q: What serializer should I use? Kryo (spark.serializer=org.apache.spark.serializer.KryoSerializer) is faster and more compact than the default Java serializer for most workloads.

Q: My YARN job can't find Hadoop config — what's wrong? Export HADOOP_CONF_DIR or YARN_CONF_DIR before submitting; Spark refuses YARN submissions without it.

13. Command Reference

bin/ — user launchers

Common environment variables (for users)

Note

Daemon/cluster environment variables (SPARK_MASTER_HOST, SPARK_WORKER_CORES, SPARK_DAEMON_MEMORY, history-server and shuffle-service options, …) are documented in the Admin Guide.