11  Introduction to PySpark and ETL Fundamentals

11.1 1. What is ETL and Why Does It Matter?

ETL (Extract, Transform, Load) is the process of moving data from operational systems (OLTP, APIs, logs, flat files) into a Data Warehouse, transforming it along the way to ensure it is clean, consistent, and ready for analytics.

Note

Distinction with Preprocessing (Part II) ETL is data engineering: it operates on production streams, terabytes of data, and fixed schemas; its output is fully automated and reproducible. Analytical preprocessing (Part II) takes a cleaned dataset view from the DW and transforms it for a specific machine learning experiment. The Data Warehouse serves as the handoff point between both workflows.

11.1.1 1.1 Classic ETL vs. Modern ELT

Classic ETL Modern ELT
Order Extract → Transform → Load Extract → Load → Transform
Execution Engine External ETL Engine Inside the DW / Lakehouse
Tools Informatica, SSIS, Talend dbt + BigQuery / Snowflake / DuckDB
Advantage Data is clean upon arrival Leverages scalable columnar DW compute
Disadvantage Potential bottleneck in ETL engine Requires scalable and cost-effective DW compute

11.1.2 1.2 Why PySpark?

When datasets exceed the RAM capacity of a single machine, pandas becomes insufficient. Apache Spark divides work into partitions processed concurrently across a cluster of nodes using an optimized execution plan (DAG — Directed Acyclic Graph).

PySpark is Python’s API for Spark. It enables writing transformations that scale seamlessly from a local machine (local mode) to hundreds of cloud nodes (AWS EMR, Databricks, GCP Dataproc) without altering code.

11.2 2. Practice Environment: Databricks Community Edition

All exercises in this section run on Databricks Community Edition, which supplies a free Spark cluster without requiring local installation.

11.2.1 2.1 Setting Up the Account

  1. Navigate to community.cloud.databricks.com and create a free account.
  2. In the sidebar: Compute → Create compute → select runtime Databricks Runtime 14.x LTS (Spark 3.5, Python 3.11).
  3. Create a new Notebook (Language: Python).
Important

In Databricks, spark is pre-instantiated as a global variable—do not instantiate a manual SparkSession. Simply begin queries with spark.read...

11.2.2 2.2 Uploading Datasets to DBFS

# Databricks File System (DBFS) — cluster file system
# Option A: Upload files via UI (File → Upload data)

# Option B: Download directly inside notebook using dbutils
dbutils.fs.mkdirs("dbfs:/datasets/taxi/")

# Download NYC Taxi dataset (Parquet, ~500 MB per month)
import urllib.request
url = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2023-01.parquet"
urllib.request.urlretrieve(url, "/tmp/yellow_2023_01.parquet")
dbutils.fs.mv("file:/tmp/yellow_2023_01.parquet", "dbfs:/datasets/taxi/yellow_2023_01.parquet")

# Verify download
display(dbutils.fs.ls("dbfs:/datasets/taxi/"))
# Read from DBFS in Databricks
df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
display(df.limit(5))   # display() is Databricks' interactive alternative to show()

11.2.3 2.3 Databricks-Specific Commands

Command Description
display(df) Renders interactive tables with built-in charting
dbutils.fs.ls("dbfs:/path/") Lists files in DBFS
dbutils.fs.mkdirs("dbfs:/path/") Creates a directory in DBFS
dbutils.fs.rm("dbfs:/path/", True) Deletes a file or directory
%sql SELECT ... Executes SQL cells against registered tables
spark.catalog.listTables() Lists available tables in the metastore
Tip

Delta Lake comes pre-installed in Databricks. There is no need for pip install delta-spark—simply pass .format("delta") directly.

11.2.4 2.4 Datasets for Exercises

Dataset Size How to obtain in Databricks
NYC Taxi (1 month) ~500 MB Download via urllib (see §2.2)
NYC Taxi (Full 2023) ~6 GB Loop across all 12 monthly URLs
Synthetic TPC-H Configurable %pip install tpch + python -m tpch generate
Databricks Sample Data Various dbfs:/databricks-datasets/ (pre-installed) 
# Explore pre-installed Databricks sample datasets
display(dbutils.fs.ls("dbfs:/databricks-datasets/"))
# Includes: nyctaxi, airlines, wine-quality, iot-stream, retail-org, etc.

# Read sample taxi dataset directly
df_sample = spark.read.format("delta")\
    .load("dbfs:/databricks-datasets/nyctaxi/tables/nyctaxi_yellow")
display(df_sample.limit(10))

11.3 3. PySpark: Core Fundamentals

11.3.1 3.1 SparkSession

In Databricks, the SparkSession is automatically exposed as spark. In local environments:

from pyspark.sql import SparkSession

# Only required outside of Databricks
spark = (SparkSession.builder
    .appName("ETL-Data-Mining")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions", "8")
    .getOrCreate())

11.3.2 3.2 Spark DataFrames: Lazy vs. Eager Execution

# Transformations (lazy) — build logical execution graph without computing results
df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
df2 = df.filter(df.trip_distance > 1.0)
df3 = df2.select("tpep_pickup_datetime", "trip_distance", "total_amount")

# Actions — trigger execution of the full DAG
df3.show(5)    # or display(df3) in Databricks
df3.count()
Concept Meaning
Transformation filter, select, groupBy, join — construct the DAG
Action show, count, collect, write — execute the DAG
Partition Parallel unit of work; typically corresponds to one file output per partition
RDD Low-level execution abstraction; use DataFrames unless handling raw custom objects

11.3.3 3.3 Explicit Schemas (Best Practice)

from pyspark.sql.types import *

schema = StructType([
    StructField("tpep_pickup_datetime",  TimestampType(), True),
    StructField("tpep_dropoff_datetime", TimestampType(), True),
    StructField("passenger_count",       IntegerType(),   True),
    StructField("trip_distance",         DoubleType(),    True),
    StructField("fare_amount",           DoubleType(),    True),
    StructField("tip_amount",            DoubleType(),    True),
    StructField("total_amount",          DoubleType(),    True),
    StructField("payment_type",          IntegerType(),   True),
])

df = spark.read.parquet("dbfs:/datasets/taxi/yellow_2023_01.parquet")
# Parquet embeds schema metadata; for CSV/JSON files, declare explicit schemas:
# df = spark.read.csv("dbfs:/path.csv", header=True, schema=schema)

Using inferSchema=True forces a full scan over raw files; specifying explicit schemas accelerates read operations and avoids unexpected type inference errors.

11.4 References for this Chapter