PySpark first steps: read a CSV, filter, aggregate, join and write Parquet

#The goal

Your team's pipelines run on Spark, and you've been asked to read or change one. You don't need a cluster to learn it. In this post you'll run PySpark on your own machine, and try each operation that shows up in nearly every job.

#Step 1: install and start

You need Python and a supported Java version:

java -version
python --version
python -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate
pip install pyspark

Check the Spark documentation for the Java versions your release supports. Create a sample file named orders.csv:

order_id,customer_id,amount,status,order_date
1,100,250.00,PAID,2026-10-01
2,101,75.50,PAID,2026-10-01
3,100,120.00,REFUNDED,2026-10-02
4,102,310.25,PAID,2026-10-02
5,,45.00,PAID,2026-10-03

And a second file, customers.csv:

customer_id,name,country
100,Asha,IN
101,Ravi,IN
102,Meera,SG

Now start a session, in a file or in the pyspark shell:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("first-steps")
         .master("local[*]")
         .getOrCreate())

local[*] runs Spark on your machine, using all of its cores. On a real cluster, you wouldn't set master in code.

#Step 2: read a file with a schema

orders = (spark.read
          .option("header", True)
          .schema("order_id INT, customer_id INT, amount DECIMAL(10,2), status STRING, order_date DATE")
          .csv("orders.csv"))

orders.printSchema()
orders.show()

Spark can guess types with inferSchema, but that reads the data an extra time, and it can guess wrong. Write the schema yourself for anything real.

Warning

By default, a CSV value that doesn't fit its type becomes NULL, with no error. You can catch it with .option("mode", "FAILFAST"), so that bad data stops the read.

#Step 3: select and filter

paid = (orders
        .filter(F.col("status") == "PAID")
        .select("order_id", "customer_id", "amount"))

paid.show()

Add a calculated column with withColumn:

with_tax = paid.withColumn("amount_with_tax", F.round(F.col("amount") * 1.18, 2))
with_tax.show()

#Step 4: understand lazy evaluation

Nothing has actually run on the data for most of those lines. Transformations such as filter and select only build a plan. Spark does the work when an action asks for a result.

See the plan with:

with_tax.explain()

#Step 5: aggregate

daily = (orders
         .filter(F.col("status") == "PAID")
         .groupBy("order_date")
         .agg(F.count("*").alias("orders"),
              F.sum("amount").alias("revenue"))
         .orderBy("order_date"))

daily.show()

Spark needs to move rows between machines to group them. That step is called a shuffle, and it's where most performance problems start. See shuffle partitions next.

#Step 6: join

customers = (spark.read.option("header", True)
             .schema("customer_id INT, name STRING, country STRING")
             .csv("customers.csv"))

joined = orders.join(customers, on="customer_id", how="left")
joined.select("order_id", "name", "country", "amount").show()

The row with no customer_id stays in a left join, with empty customer columns. With an inner join, it would disappear. Always compare row counts before and after a join:

print(orders.count(), joined.count())

If the count goes up, read about join row explosions.

#Step 7: write Parquet

(joined
 .write
 .mode("overwrite")
 .partitionBy("order_date")
 .parquet("output/orders_enriched"))

Save modes:

Caution

overwrite removes the whole target path by default, even with partitionBy. That's a common way to lose data. See overwrite deleted all partitions before you use it on real data.

Read the output back to check it:

check = spark.read.parquet("output/orders_enriched")
check.filter(F.col("order_date") == "2026-10-02").show()

Filtering on a partition column means Spark reads only those folders. For more on why Parquet is a good fit, read file formats.

#Step 8: use SQL if you prefer it

orders.createOrReplaceTempView("orders")

spark.sql("""
  SELECT status, COUNT(*) AS n, SUM(amount) AS total
  FROM orders
  GROUP BY status
""").show()

SQL and the DataFrame API run on the same engine, so choose the one that's easier to read.

#Common errors

Error names vary a little between Spark versions.

#Don't pull everything to the driver

collect() and toPandas() bring all the rows into a single process. On a real dataset, that crashes the driver. Use them only on results that you know are small:

top = orders.orderBy(F.desc("amount")).limit(10).toPandas()

#Stop the session

spark.stop()

#Prevent it next time

• Write schemas, rather than inferring them.
• Compare row counts around each join.
• Open the Spark UI at http://localhost:4040 while a job is running. The Spark UI guide explains what to look for.
• When you're ready for tuning settings, see the spark-submit cheat sheet.