PySpark Tutorial: RDDs, DataFrames, and Spark SQL in Python

What Is PySpark?

PySpark is the Python API for Apache Spark. Your Python code describes the transformations you want, and Spark's engine, which runs on the Java Virtual Machine (JVM), plans and executes them across the cluster. You get Spark's distributed execution without writing Scala or Java.

Data scientists use PySpark to clean and aggregate large datasets, data engineers use it for ETL pipelines, and both use pyspark.ml to train models on data that won't fit on one machine. If most of your work happens in pandas today, the DataFrame API will feel familiar.

When to use PySpark instead of pandas

PySpark pays off when data outgrows a single machine; below that, pandas is usually faster because it has no JVM startup or task scheduling overhead.

My rule of thumb: if the data fits in memory with room to spare and the job runs on one laptop, stay with pandas. Switch to PySpark when you're fighting memory errors, or when the job will eventually run on a cluster or a platform like Databricks.

How to Install PySpark

PySpark installs from PyPI with one pip command, which bundles the Spark runtime, so you don't need a separate Spark or Hadoop download for local work. Before running it, check the two prerequisites.

Prerequisites

According to the official installation guide, PySpark 4.3 requires:

• Python 3.11 or later. PySpark 4.3 dropped Python 3.10; see our Python setup guide if you need a newer version.
• Java 17 or later (Spark 4.3 runs on Java 17, 21, and 25) with the JAVA_HOME environment variable pointing to it. Our guide to installing Java walks through each operating system.

You can check both from a terminal:

python3 --version  # Python 3.11 or later
java -version      # openjdk version "17" or later제공

If you work in DataLab or Databricks, PySpark is already installed, and you can skip to the dataset.

Installing PySpark with pip

I recommend a virtual environment so PySpark's dependencies don't collide with other projects:

python3 -m venv pyspark-env
source pyspark-env/bin/activate
pip install "pyspark[sql]" "pandas<3"
python3 -c "import pyspark; print(pyspark.__version__)"제공

4.3.0제공

On Windows, use python instead of python3, and activate the environment with pyspark-env\Scripts\activate. The [sql] extra adds pandas, PyArrow, and NumPy, which PySpark uses for toPandas() and pandas UDFs. I pin pandas<3 because PySpark 4.3 prints a warning that it doesn't yet fully support pandas 3.0.

Two errors account for most failed first runs:

• JAVA_GATEWAY_EXITED or "Unable to locate a Java Runtime": Java is missing or JAVA_HOME isn't set.
• PYTHON_VERSION_MISMATCH: Spark's worker processes picked up a different Python than your script. Set PYSPARK_PYTHON to your environment's interpreter, for example export PYSPARK_PYTHON=$(which python).

For operating-system-specific steps, including Windows, see our PySpark installation guide.

Getting the dataset

The examples use an e-commerce dataset of 541,909 transactions from a UK-based online retailer, covering December 2010 to December 2011. Each row is one product line on an invoice, with 8 columns:

• InvoiceNo: invoice number (numbers starting with "C" are cancellations)
• StockCode and Description: product code and name
• Quantity and UnitPrice: units bought and price per unit, in pounds sterling
• InvoiceDate: date and time of the purchase
• CustomerID: customer identifier (missing on some rows)
• Country: the customer's country

You can download it as datacamp_ecommerce.csv from the DataLab dataset page. It's the Online Retail dataset from the UCI Machine Learning Repository, so you can also build the same file from the public source (requires pip install openpyxl):

import io
import urllib.request
import zipfile

import pandas as pd

url = "https://archive.ics.uci.edu/static/public/352/online+retail.zip"
with urllib.request.urlopen(url) as response:
    archive = zipfile.ZipFile(io.BytesIO(response.read()))

retail = pd.read_excel(
    archive.open("Online Retail.xlsx"),
    dtype={"InvoiceNo": str, "StockCode": str, "CustomerID": "Int64"},
)
# Match the DataLab file's date format, e.g. 12/1/10 8:26
retail["InvoiceDate"] = retail["InvoiceDate"].map(
    lambda ts: f"{ts.month}/{ts.day}/{ts:%y} {ts.hour}:{ts:%M}"
)
retail.to_csv("datacamp_ecommerce.csv", index=False)제공

Reading the 23 MB Excel file takes about a minute. Save the CSV in the folder you'll run the rest of the code from.

Creating a SparkSession

A SparkSession is the entry point to every PySpark program: it connects your Python code to the Spark engine and creates DataFrames. You create one per application with the builder pattern:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PySparkTutorial").getOrCreate()
print(spark.version)제공

4.3.0제공

getOrCreate() returns the running session if one exists, so it's safe to call twice in a notebook. Launched from a plain Python process, Spark runs in local mode (local[*]), using every CPU core on your machine as a worker, and serves the Spark UI at http://localhost:4040 for watching jobs. When you're done, spark.stop() releases the resources.

Loading the e-commerce dataset

With the session running, we can read the CSV into a DataFrame. I pass an explicit schema rather than inferSchema=True, which makes Spark scan the whole file an extra time just to guess column types:

import pyspark.sql.functions as F

schema = """
    InvoiceNo STRING, StockCode STRING, Description STRING, Quantity INT,
    InvoiceDate STRING, UnitPrice DOUBLE, CustomerID INT, Country STRING
"""

orders = spark.read.csv(
    "datacamp_ecommerce.csv", header=True, schema=schema, escape='"'
)
orders = orders.withColumn("InvoiceDate", F.to_timestamp("InvoiceDate", "M/d/yy H:mm"))

orders.printSchema()제공

root
 |-- InvoiceNo: string (nullable = true)
 |-- StockCode: string (nullable = true)
 |-- Description: string (nullable = true)
 |-- Quantity: integer (nullable = true)
 |-- InvoiceDate: timestamp (nullable = true)
 |-- UnitPrice: double (nullable = true)
 |-- CustomerID: integer (nullable = true)
 |-- Country: string (nullable = true)제공

Two details in that snippet matter:

• escape='"' handles product names that contain doubled quotes, such as RECORD FRAME 7"" SINGLE SIZE. Spark's default escape character is a backslash.
• The date pattern M/d/yy H:mm matches values like 12/1/10 8:26. Since Spark 4.0, ANSI SQL mode is on by default, so to_timestamp() raises a CANNOT_PARSE_TIMESTAMP error on a value that doesn't match the pattern. Use F.try_to_timestamp() if you'd rather get NULL for bad rows.

To look at the first rows and count them:

orders.show(5, truncate=False)
print(orders.count())제공

+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
|InvoiceNo|StockCode|Description                        |Quantity|InvoiceDate        |UnitPrice|CustomerID|Country       |
+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
|536365   |85123A   |WHITE HANGING HEART T-LIGHT HOLDER |6       |2010-12-01 08:26:00|2.55     |17850     |United Kingdom|
|536365   |71053    |WHITE METAL LANTERN                |6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
|536365   |84406B   |CREAM CUPID HEARTS COAT HANGER     |8       |2010-12-01 08:26:00|2.75     |17850     |United Kingdom|
|536365   |84029G   |KNITTED UNION FLAG HOT WATER BOTTLE|6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
|536365   |84029E   |RED WOOLLY HOTTIE WHITE HEART.     |6       |2010-12-01 08:26:00|3.39     |17850     |United Kingdom|
+---------+---------+-----------------------------------+--------+-------------------+---------+----------+--------------+
only showing top 5 rows

541909제공

For more reading options, such as multiple files, delimiters, and corrupt records, see our PySpark read CSV tutorial.

RDDs vs. DataFrames

RDDs and DataFrames are PySpark's two ways to represent distributed data, and for almost all new code, DataFrames are the right choice.

Here's the same aggregation, total quantity per country, written both ways:

sales = [("United Kingdom", 6), ("France", 24), ("United Kingdom", 8)]

# RDD: you write the logic with Python functions
rdd = spark.sparkContext.parallelize(sales)
print(rdd.reduceByKey(lambda a, b: a + b).collect())

# DataFrame: you declare the result, and Spark plans how to compute it
sales_df = spark.createDataFrame(sales, ["country", "quantity"])
sales_df.groupBy("country").sum("quantity").show()제공

[('France', 24), ('United Kingdom', 14)]
+--------------+-------------+
|       country|sum(quantity)|
+--------------+-------------+
|United Kingdom|           14|
|        France|           24|
+--------------+-------------+제공

The RDD version runs your lambda in Python worker processes. The DataFrame version goes through Spark's Catalyst optimizer, which rewrites the query into an efficient plan that runs inside the JVM, so DataFrame code is usually faster without any tuning. The rest of this tutorial uses DataFrames.

Lazy Evaluation: Transformations vs. Actions

Lazy evaluation means Spark doesn't execute a transformation when you call it; it records the step in a plan and runs the whole plan only when an action asks for a result.

The filter below returns instantly, even though the dataset has over half a million rows, because nothing is read yet. The count() call is what launches a job:

uk_orders = orders.filter(F.col("Country") == "United Kingdom")  # transformation: no job yet
print(uk_orders.count())                                          # action: Spark reads the file now제공

495478제공

Waiting until an action lets Spark optimize the whole chain at once. explain() prints the physical plan it chose (trimmed here):

uk_orders.explain()제공

== Physical Plan ==
*(1) Project [InvoiceNo#0, StockCode#1, ...]
+- *(1) Filter (isnotnull(Country#7) AND (Country#7 = United Kingdom))
   +- FileScan csv [...] PushedFilters: [IsNotNull(Country), EqualTo(Country,United Kingdom)], ...제공

The PushedFilters entry shows Spark pushing the country filter down into the file scan, so rows from other countries are dropped as the CSV is read. One practical consequence: a data error inside a transformation, such as a date that fails to parse under ANSI mode, surfaces only when the next action runs, several lines later.

Common DataFrame Operations

Most PySpark code is a chain of five DataFrame operations: select(), filter(), withColumn(), groupBy(), and join(). Each returns a new DataFrame, so you can chain them in any order.

Selecting columns with select()

select() returns a DataFrame with only the columns you name:

orders.select("InvoiceNo", "Description", "Quantity", "UnitPrice").show(3, truncate=False)제공

+---------+----------------------------------+--------+---------+
|InvoiceNo|Description                       |Quantity|UnitPrice|
+---------+----------------------------------+--------+---------+
|536365   |WHITE HANGING HEART T-LIGHT HOLDER|6       |2.55     |
|536365   |WHITE METAL LANTERN               |6       |3.39     |
|536365   |CREAM CUPID HEARTS COAT HANGER    |8       |2.75     |
+---------+----------------------------------+--------+---------+
only showing top 3 rows제공

Filtering rows with filter()

filter() keeps the rows that match a condition. This dataset mixes sales with cancellations (invoice numbers starting with "C"), negative quantities, and rows without a customer, so I'll keep only valid sales:

sales_only = orders.filter(
    (F.col("Quantity") > 0)
    & (F.col("UnitPrice") > 0)
    & F.col("CustomerID").isNotNull()
    & ~F.col("InvoiceNo").startswith("C")
)
print(sales_only.count())제공

397884제공

Combine conditions with & (and), | (or), and ~ (not), and wrap each one in parentheses, because Python's operator precedence would otherwise evaluate & before >. Our PySpark filter tutorial covers more patterns.

Adding columns with withColumn()

withColumn() adds a column, or replaces one with the same name. Here it calculates the value of each line:

sales_only = sales_only.withColumn("TotalAmount", F.round(F.col("Quantity") * F.col("UnitPrice"), 2))
sales_only.select("InvoiceNo", "Quantity", "UnitPrice", "TotalAmount").show(3)제공

+---------+--------+---------+-----------+
|InvoiceNo|Quantity|UnitPrice|TotalAmount|
+---------+--------+---------+-----------+
|   536365|       6|     2.55|       15.3|
|   536365|       6|     3.39|      20.34|
|   536365|       8|     2.75|       22.0|
+---------+--------+---------+-----------+
only showing top 3 rows제공

See our PySpark withColumn() tutorial for conditional columns with F.when().

Aggregating with groupBy() and agg()

groupBy() groups rows by one or more columns, and agg() computes one or more summaries per group. Which countries bring in the most revenue?

country_revenue = (
    sales_only.groupBy("Country")
    .agg(
        F.count_distinct("CustomerID").alias("customers"),
        F.round(F.sum("TotalAmount"), 2).alias("revenue"),
    )
    .orderBy(F.desc("revenue"))
)
country_revenue.show(5)제공

+--------------+---------+----------+
|       Country|customers|   revenue|
+--------------+---------+----------+
|United Kingdom|     3920|7308391.55|
|   Netherlands|        9| 285446.34|
|          EIRE|        3|  265545.9|
|       Germany|       94| 228867.14|
|        France|       87| 209024.05|
+--------------+---------+----------+
only showing top 5 rows제공

The UK dominates, as you'd expect from a UK retailer. The more interesting rows are the Netherlands and Ireland (listed as EIRE): 9 and 3 customers spend more than Germany's 94, which suggests wholesale buyers.

Notice I import the functions module as F instead of using from pyspark.sql.functions import *, which shadows Python's built-in sum(), min(), and max(). Since PySpark 4.0, that wildcard import also no longer exposes the old camelCase name countDistinct.

Our PySpark groupBy tutorial goes deeper into aggregations.

Joining DataFrames with join()

join() combines two DataFrames on a shared key. To label each country with a region, I'll join a small lookup table that deliberately leaves out Australia:

regions = spark.createDataFrame(
    [("United Kingdom", "UK"), ("Netherlands", "EU"), ("EIRE", "EU"),
     ("Germany", "EU"), ("France", "EU")],
    ["Country", "region"],
)
country_revenue.join(regions, on="Country", how="left").orderBy(F.desc("revenue")).show(6)제공

+--------------+---------+----------+------+
|       Country|customers|   revenue|region|
+--------------+---------+----------+------+
|United Kingdom|     3920|7308391.55|    UK|
|   Netherlands|        9| 285446.34|    EU|
|          EIRE|        3|  265545.9|    EU|
|       Germany|       94| 228867.14|    EU|
|        France|       87| 209024.05|    EU|
|     Australia|        9| 138521.31|  NULL|
+--------------+---------+----------+------+
only showing top 6 rows제공

A left join keeps every row from the left DataFrame and fills missing matches with NULL, which is why Australia stays in the result. An inner join (how="inner", the default) would drop it. Our PySpark joins tutorial compares all join types and their performance.

Spark SQL in PySpark

Spark SQL lets you query a DataFrame with standard SQL by registering it as a temporary view. This query reproduces the country revenue table from the groupBy section:

sales_only.createOrReplaceTempView("sales")

spark.sql("""
    SELECT Country,
           COUNT(DISTINCT CustomerID) AS customers,
           ROUND(SUM(TotalAmount), 2) AS revenue
    FROM sales
    GROUP BY Country
    ORDER BY revenue DESC
    LIMIT 5
""").show()제공

+--------------+---------+----------+
|       Country|customers|   revenue|
+--------------+---------+----------+
|United Kingdom|     3920|7308391.55|
|   Netherlands|        9| 285446.34|
|          EIRE|        3|  265545.9|
|       Germany|       94| 228867.14|
|        France|       87| 209024.05|
+--------------+---------+----------+제공

Both versions compile to the same optimized plan, so neither is faster. I use SQL for long aggregations that analysts will read, and the DataFrame API when I need Python logic around the query, such as building filters in a loop. spark.sql() returns a DataFrame, so you can mix the two freely.

The temporary view lasts until the SparkSession stops.

Machine Learning with PySpark: A K-Means Example

PySpark's pyspark.ml module trains models on DataFrames, and a Pipeline chains feature preparation and the model into one object you can fit and reuse. To show it, I'll segment the retailer's customers with K-Means clustering on recency, frequency, and monetary value (RFM), a standard marketing segmentation:

• Recency: days since the customer's last purchase
• Frequency: number of distinct invoices
• Monetary: total amount spent

Building RFM features with groupBy()

One groupBy() over the cleaned sales_only DataFrame produces all 3 features. Recency is measured from the day after the last invoice in the data:

snapshot = sales_only.agg(F.date_add(F.max("InvoiceDate"), 1)).first()[0]

rfm = sales_only.groupBy("CustomerID").agg(
    F.datediff(F.lit(snapshot), F.max("InvoiceDate")).alias("recency"),
    F.count_distinct("InvoiceNo").alias("frequency"),
    F.round(F.sum("TotalAmount"), 2).alias("monetary"),
)
rfm.describe("recency", "frequency", "monetary").show()제공

+-------+------------------+-----------------+------------------+
|summary|           recency|        frequency|          monetary|
+-------+------------------+-----------------+------------------+
|  count|              4338|             4338|              4338|
|   mean| 93.05947441217151|4.272014753342554|2054.2664591977896|
| stddev|100.01226424171826|7.697997826095038| 8989.230441421178|
|    min|                 1|                1|              3.75|
|    max|               374|              209|         280206.02|
+-------+------------------+-----------------+------------------+제공

Look at the monetary column: a mean of about £2,054, a standard deviation over four times that, and a maximum above £280,000. K-Means measures Euclidean distance, so a few big spenders would dominate the clusters. Taking the log of each feature first compresses that long tail.

Scaling and clustering in a Pipeline

The Pipeline has 3 stages: VectorAssembler packs the features into the single vector column that pyspark.ml models expect, StandardScaler puts them on the same scale, and KMeans clusters them. I fit it for k = 2 to 6 and score each model with the silhouette score, which runs from -1 to 1, higher meaning better-separated clusters:

from pyspark.ml import Pipeline
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
from pyspark.ml.feature import StandardScaler, VectorAssembler

rfm_log = rfm.select(
    "CustomerID",
    "recency", "frequency", "monetary",
    F.log1p("recency").alias("log_recency"),
    F.log1p("frequency").alias("log_frequency"),
    F.log1p("monetary").alias("log_monetary"),
).cache()

assembler = VectorAssembler(
    inputCols=["log_recency", "log_frequency", "log_monetary"], outputCol="features_raw"
)
scaler = StandardScaler(
    inputCol="features_raw", outputCol="features", withMean=True, withStd=True
)
evaluator = ClusteringEvaluator(featuresCol="features", metricName="silhouette")

for k in range(2, 7):
    kmeans = KMeans(featuresCol="features", k=k, seed=42)
    model = Pipeline(stages=[assembler, scaler, kmeans]).fit(rfm_log)
    score = evaluator.evaluate(model.transform(rfm_log))
    print(f"k={k}  silhouette={score:.3f}")제공

k=2  silhouette=0.624
k=3  silhouette=0.502
k=4  silhouette=0.502
k=5  silhouette=0.478
k=6  silhouette=0.470제공

.cache() keeps rfm_log in memory, because every fit and evaluation in the loop reads it again; without it, Spark would recompute the RFM table from the CSV on every fit.

Choosing k and reading the segments

The silhouette score peaks at k = 2, but 2 segments mostly split active customers from lapsed ones, which a marketing team can't do much with. k = 4 scores the same as k = 3 and splits the active customers into big spenders and newcomers, so I'll use it and profile each cluster:

kmeans = KMeans(featuresCol="features", k=4, seed=42)
model = Pipeline(stages=[assembler, scaler, kmeans]).fit(rfm_log)
segments = model.transform(rfm_log)

(segments.groupBy("prediction")
    .agg(
        F.count("*").alias("customers"),
        F.round(F.avg("recency"), 1).alias("avg_recency"),
        F.round(F.avg("frequency"), 1).alias("avg_frequency"),
        F.round(F.avg("monetary"), 1).alias("avg_monetary"),
    )
    .orderBy("prediction")
    .show())제공

+----------+---------+-----------+-------------+------------+
|prediction|customers|avg_recency|avg_frequency|avg_monetary|
+----------+---------+-----------+-------------+------------+
|         0|     1598|      181.2|          1.3|       326.2|
|         1|      720|       13.2|         13.7|      8072.7|
|         2|     1162|       77.1|          3.9|      1765.2|
|         3|      858|       17.4|          2.3|       613.7|
+----------+---------+-----------+-------------+------------+제공

The four segments read cleanly:

• Cluster 1, best customers (720): bought within the last 2 weeks on average, with about 14 orders and over £8,000 spent each. Retention and loyalty offers go here.
• Cluster 3, recent and light (858): active in the last 3 weeks but with only 2 or so orders. Many are likely new customers worth nurturing toward a second and third order.
• Cluster 2, cooling off (1,162): mid-range spend, but about 11 weeks since their last purchase. A win-back campaign has the best odds here.
• Cluster 0, lapsed one-time buyers (1,598): mostly a single order, around £326, roughly 6 months ago. The largest group and the least valuable.

Cluster numbers are arbitrary labels, and a different seed can reorder them, so always profile the clusters before naming them. To learn how K-Means picks its centroids, see our introduction to k-means clustering, and the pyspark.ml API reference lists every model and transformer. Our Machine Learning with PySpark course covers classification, regression, and model tuning with Pipelines.

Final Thoughts

PySpark comes down to a few ideas: a SparkSession to start, DataFrames for almost everything, lazy transformations that run only when an action needs a result, and the choice of writing each query in Python or SQL. Everything in this tutorial ran in local mode on a laptop, and the same code runs unchanged on a cluster.

A 541,909-row file doesn't need a cluster, and pandas would have handled it. That was deliberate, because the habits that matter on large data (explicit schemas, filtering early, caching what you reuse, and checking explain()) are easier to build on data you can inspect.

From here, these are the best next steps, depending on your goal:

• Practice the DataFrame API with exercises: Introduction to PySpark
• Prepare messy real-world data: Cleaning Data with PySpark
• Keep the syntax at hand: our PySpark cheat sheet
• Prepare for a job interview: Top PySpark interview questions

If you're keen to go further, I recommend our Big Data with PySpark track, which covers data management, feature engineering, machine learning pipelines, and recommendation engines in sequence. For teams, DataCamp for Business offers learning paths for data engineering roles.