PySpark in Ten Minutes: Your First DataFrame
Install PySpark, start a SparkSession, build a DataFrame, and see why nothing runs until you call an action.
PySpark is a Python API over a distributed execution engine. The surprising part for newcomers is not the distribution — it is that your code describes work rather than doing it. This lesson makes that visible.
Installing
pip install pyspark==3.5.4Successfully installed py4j-0.10.9.7 pyspark-3.5.4PySpark bundles Spark itself, so there is nothing else to install. You do need a JVM:
java -versionopenjdk version "17.0.13" 2024-10-15
OpenJDK Runtime Environment (build 17.0.13+11)
OpenJDK 64-Bit Server VM (build 17.0.13+11, mixed mode, sharing)Starting a session
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("first-steps")
.master("local[*]") # use every core on this machine as an executor
.getOrCreate()
)
print(spark.version)
print(spark.sparkContext.defaultParallelism, "default partitions")$ python first_session.py
25/02/14 11:20:03 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform
Setting default log level to "WARN".
3.5.4
8 default partitionsThe NativeCodeLoader warning is normal on machines without Hadoop’s native libraries and
has no effect on correctness — ignore it.
local[*] means “run locally, using all cores”. On a real cluster this becomes a YARN or
Kubernetes URL and nothing else in your code changes.
Building a DataFrame
data = [
("alice", "engineering", 95000, 2019),
("bob", "engineering", 87000, 2021),
("carol", "design", 78000, 2020),
("dave", "design", 82000, 2018),
("erin", "sales", 67000, 2022),
]
df = spark.createDataFrame(data, ["name", "team", "salary", "joined"])
df.show()$ python build_df.py
+-----+-----------+------+------+
| name| team|salary|joined|
+-----+-----------+------+------+
|alice|engineering| 95000| 2019|
| bob|engineering| 87000| 2021|
|carol| design| 78000| 2020|
| dave| design| 82000| 2018|
| erin| sales| 67000| 2022|
+-----+-----------+------+------+Spark inferred the types from the Python values:
df.printSchema()root
|-- name: string (nullable = true)
|-- team: string (nullable = true)
|-- salary: long (nullable = true)
|-- joined: long (nullable = true)Note long, not int — Python integers become 64-bit in Spark. Inference is fine for small
literal data; for files you should always declare the schema, which the
reading and writing lesson covers.
Nothing has run yet
This is the idea that makes Spark make sense. Time a filter:
import time
start = time.perf_counter()
senior = df.filter(df.salary > 80000)
print(f"filter returned in {time.perf_counter() - start:.6f}s")
start = time.perf_counter()
senior.show()
print(f"show took {time.perf_counter() - start:.3f}s")filter returned in 0.001820s
+-----+-----------+------+------+
| name| team|salary|joined|
+-----+-----------+------+------+
|alice|engineering| 95000| 2019|
| bob|engineering| 87000| 2021|
| dave| design| 82000| 2018|
+-----+-----------+------+------+
show took 0.847sThe filter took under two milliseconds because it did no work. It returned a new DataFrame
carrying a description of the filter. Only show() — an action — caused Spark to plan
and execute.
That laziness is what lets Spark optimise. Chain several transformations and Spark still sees the whole thing before running any of it:
result = (
df.filter(df.salary > 70000)
.filter(df.joined < 2021)
.select("name", "salary")
)
result.explain()== Physical Plan ==
*(1) Project [name#0, salary#2L]
+- *(1) Filter ((isnotnull(salary#2L) AND (salary#2L > 70000)) AND (joined#3L < 2021))
+- *(1) Scan ExistingRDD[name#0,team#1,salary#2L,joined#3L]Read it bottom-up. Two separate filter calls became one Filter node with both
conditions combined, and Spark added an isnotnull check of its own. It also pushed the
projection down so only name and salary are carried forward. You wrote three steps;
Spark decided to run one.
Errors surface at the action, not the transformation
A consequence of laziness that catches everyone once:
broken = df.select("name", "salary", "department") # no such column
print("transformation accepted")Traceback (most recent call last):
...
pyspark.errors.exceptions.captured.AnalysisException:
[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name
`department` cannot be resolved. Did you mean one of the following? [`team`, `name`, `salary`, `joined`].select fails immediately because Spark checks column names against the known schema at
analysis time. But a runtime error inside a UDF, or a malformed file, will not appear until
an action runs — possibly minutes into a job. When a Spark traceback points at a line far
from the real problem, this is usually why.
Actions worth knowing
print("count: ", df.count())
print("first row: ", df.first())
print("as python: ", df.select("name").limit(2).collect())count: 5
first row: Row(name='alice', team='engineering', salary=95000, joined=2019)
as python: [Row(name='alice'), Row(name='bob')]collect() pulls every row into the driver’s memory. On five rows that is fine; on a
billion-row DataFrame it will crash the driver. Use limit() before collect(), or show(),
or write to storage — reaching for collect() on a large DataFrame is the most common way to
kill a Spark job.
Always stop the session when you are done:
spark.stop()Practice
1. Create a DataFrame of five books with title, author, and year. Show only books published after 2000.
books = spark.createDataFrame(
[("Dune", "Herbert", 1965), ("Neuromancer", "Gibson", 1984),
("Cloud Atlas", "Mitchell", 2004), ("The Road", "McCarthy", 2006),
("Piranesi", "Clarke", 2020)],
["title", "author", "year"])
books.filter(books.year > 2000).show()+-----------+--------+----+
| title| author|year|
+-----------+--------+----+
|Cloud Atlas|Mitchell|2004|
| The Road|McCarthy|2006|
| Piranesi| Clarke|2020|
+-----------+--------+----+2. Call explain() on a chain of four filters. How many Filter nodes appear?
== Physical Plan ==
*(1) Filter (((((isnotnull(salary#2L) AND isnotnull(joined#3L)) AND (salary#2L > 60000))
AND (salary#2L < 100000)) AND (joined#3L > 2017)) AND (joined#3L < 2023))
+- *(1) Scan ExistingRDD[...]One. Catalyst collapses consecutive filters into a single node, so writing them as separate readable steps costs nothing at runtime. Optimise for clarity in the source; Spark will optimise the execution.
3. What does df.filter(df.salary > 80000) return — rows, or something else?
print(type(df.filter(df.salary > 80000)))<class 'pyspark.sql.dataframe.DataFrame'>A new DataFrame. Transformations always return DataFrames and never row data; only actions return values to Python. Getting this distinction right is most of learning Spark.
4. Run df.count() twice and time both. Why is the second not faster?
first count: 0.412s
second count: 0.388sRoughly the same, because Spark recomputes the whole lineage every time unless you tell it
not to. Add df.cache() and the second call drops sharply:
first count: 0.445s
second count: 0.021sCaching is covered with the execution model; for now, know that DataFrames are recipes, not results.
Next: building DataFrames from real files, with schemas you control.