How to Build a Simple ETL Job Locally with PySpark, PostgreSQL, and Docker

Originally published on itnext.io. This article demonstrates how Apache Spark can power Python-based ETL jobs, using a real-world used-car dataset scraped from Ebay-Kleinanzeigen (the German Ebay classifieds platform, containing data for over 370,000 used cars).

Introduction

If you work with data day-to-day in Python and want to move toward more scalable processing, PySpark is a natural next step. This walkthrough shows how to build a complete, locally runnable ETL pipeline that:

• Extracts raw CSV data into a Spark DataFrame
• Transforms the data by cleaning and dropping irrelevant or empty columns
• Loads the cleaned data into a PostgreSQL database running inside Docker

What Is Apache Spark?

Apache Spark is an open-source, distributed data processing engine that supports multiple programming languages, including Python (via PySpark), Java, Scala, and R. Its key advantage over older batch frameworks such as Hadoop MapReduce is that it processes data in memory, which dramatically reduces read/write overhead to disk. Spark ships with built-in libraries for SQL queries, streaming, machine learning (MLlib), and graph processing (GraphX). It can run on a laptop in local mode or be scaled to a cluster of hundreds of nodes.

What Is ETL?

ETL stands for Extract, Transform, Load—the standard pattern for moving data between systems:

• Extract — Read raw data from one or more source systems (files, APIs, databases).
• Transform — Clean, validate, reshape, and enrich the data so it is fit for analysis.
• Load — Write the processed data to a target store such as a relational database, data warehouse, or data lake.

Prerequisites

• Python 3.x with pyspark and psycopg2 installed
• Docker and Docker Compose installed locally
• The Ebay-Kleinanzeigen used-car CSV dataset (available on Kaggle)

Step 1 — Extract: Reading CSV Data with PySpark

Creating a SparkSession

Every PySpark application begins with a SparkSession. This is the driver-side entry point that tracks application state, schedules tasks, and coordinates executors. In local mode the entire job runs on your machine:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[*]")        # use all available CPU cores locally
    .appName("UsedCarsETL")
    .getOrCreate()
)Code language: PHP (php)

• local[*] — Spark spins up as many worker threads as there are logical CPU cores on the host machine.
• getOrCreate() — Returns an existing session if one is already active, otherwise creates a new one. Useful when running inside a notebook or when re-running a script.
• appName — Labels the job in the Spark UI (accessible at http://localhost:4040 while the job runs).

Reading the CSV with Schema Inference

The quickest way to load a CSV is to let Spark infer the schema automatically:

df = (
    spark.read
    .option("header", True)
    .option("inferSchema", True)
    .csv("autos.csv")
)Code language: PHP (php)

Schema inference is convenient for exploration, but it has two important drawbacks in production:

• It requires Spark to perform an extra scan of the data to guess column types, which slows down start-up.
• Inferred types may not match your expectations—numeric-looking strings can be cast incorrectly.

Reading with an Explicit Schema (Recommended for Production)

Define a schema up front using StructType and StructField. This skips the inference scan and guarantees column types:

from pyspark.sql.types import (
    StructType, StructField,
    StringType, IntegerType, DoubleType, TimestampType
)

schema = StructType([
    StructField("dateCrawled",        TimestampType(), True),
    StructField("name",               StringType(),    True),
    StructField("seller",             StringType(),    True),
    StructField("offerType",          StringType(),    True),
    StructField("price",              IntegerType(),   True),
    StructField("abtest",             StringType(),    True),
    StructField("vehicleType",        StringType(),    True),
    StructField("yearOfRegistration", IntegerType(),   True),
    StructField("gearbox",            StringType(),    True),
    StructField("powerPS",            IntegerType(),   True),
    StructField("model",              StringType(),    True),
    StructField("odometer",           IntegerType(),   True),
    StructField("monthOfRegistration",IntegerType(),   True),
    StructField("fuelType",           StringType(),    True),
    StructField("brand",              StringType(),    True),
    StructField("notRepairedDamage",  StringType(),    True),
    StructField("dateCreated",        TimestampType(), True),
    StructField("nrOfPictures",       IntegerType(),   True),
    StructField("postalCode",         IntegerType(),   True),
    StructField("lastSeen",           TimestampType(), True),
])

df = (
    spark.read
    .option("header", True)
    .schema(schema)
    .csv("autos.csv")
)Code language: PHP (php)

Step 2 — Transform: Cleaning the Dataset

After loading the data you can inspect it with df.show() and df.printSchema(). The dataset contains several columns that are not useful for downstream analysis:

• dateCrawled and lastSeen — crawler metadata, not analytically relevant.
• nrOfPictures — every value in this column is 0, making it useless.
• seller — almost all rows (371,525 out of ~371,528) are labelled privat; the three gewerblich rows can be filtered out and the column dropped.
• offerType — similarly, only 12 rows carry the value Gesuch versus 371,513 rows labelled Angebot.

# Drop columns with no analytical value
cols_to_drop = ["dateCrawled", "lastSeen", "nrOfPictures"]
df_clean = df.drop(*cols_to_drop)

# Remove the tiny minority of commercial sellers and requests
df_clean = (
    df_clean
    .filter(df_clean.seller    == "privat")
    .filter(df_clean.offerType == "Angebot")
    .drop("seller", "offerType")
)Code language: PHP (php)

Null values remain in the dataset. Strategies such as imputation, median fill, or row removal depend on the analytical goal and are beyond the scope of this walkthrough; here the nulls are left as-is and handled at query time.

Step 3 — Spin Up PostgreSQL with Docker Compose

Instead of installing PostgreSQL locally, use Docker Compose to start a Postgres instance and pgAdmin together. Create a docker-compose.yml file:

version: "3.7"
services:
  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB:       cars
      POSTGRES_USER:     admin
      POSTGRES_PASSWORD: admin
    ports:
      - "5432:5432"

  pgadmin:
    image: dpage/pgadmin4
    environment:
      PGADMIN_DEFAULT_EMAIL:    admin@admin.com
      PGADMIN_DEFAULT_PASSWORD: admin
    ports:
      - "5050:80"
    depends_on:
      - postgresCode language: JavaScript (javascript)

Start the services with:

docker-compose up -d

PostgreSQL is now reachable at localhost:5432 and pgAdmin at http://localhost:5050.

Step 4 — Load: Writing Data to PostgreSQL with Psycopg2

Establishing a Connection

Psycopg2 is the most widely used PostgreSQL adapter for Python. Connect to the cars database created by Docker Compose:

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="cars",
    user="admin",
    password="admin"
    # port defaults to 5432
)Code language: PHP (php)

Creating a Cursor and the Target Table

A cursor is the object through which you send SQL commands to PostgreSQL:

cur = conn.cursor()

create_table_query = """
CREATE TABLE IF NOT EXISTS cars_table (
    name                 TEXT,
    price                INTEGER,
    vehicle_type         TEXT,
    year_of_registration INTEGER,
    gearbox              TEXT,
    power_ps             INTEGER,
    model                TEXT,
    odometer             INTEGER,
    month_of_registration INTEGER,
    fuel_type            TEXT,
    brand                TEXT,
    not_repaired_damage  TEXT,
    date_created         TIMESTAMP,
    postal_code          INTEGER
);
"""

cur.execute(create_table_query)Code language: PHP (php)

Inserting Rows

Convert the Spark DataFrame to a list of tuples and use executemany for a bulk insert:

# Collect the cleaned Spark DataFrame to the driver as a list of Row objects
cars_rows = df_clean.collect()

# Convert to a list of plain Python tuples
cars_seq = [
    (
        row.name, row.price, row.vehicleType,
        row.yearOfRegistration, row.gearbox, row.powerPS,
        row.model, row.odometer, row.monthOfRegistration,
        row.fuelType, row.brand, row.notRepairedDamage,
        row.dateCreated, row.postalCode
    )
    for row in cars_rows
]

insert_query = """
INSERT INTO cars_table (
    name, price, vehicle_type, year_of_registration,
    gearbox, power_ps, model, odometer,
    month_of_registration, fuel_type, brand,
    not_repaired_damage, date_created, postal_code
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""

cur.executemany(insert_query, cars_seq)Code language: PHP (php)

Important: Psycopg2 is Python DB-API 2.0 compliant, so auto-commit is disabled by default. You must explicitly commit the transaction, otherwise your inserts will be rolled back when the connection closes:

conn.commit()
cur.close()
conn.close()Code language: CSS (css)

Verifying the Load

You can spot-check the loaded data with a simple query:

conn = psycopg2.connect(host="localhost", database="cars",
                        user="admin", password="admin")
cur = conn.cursor()

cur.execute("SELECT brand, model, price FROM cars_table LIMIT 2;")
rows = cur.fetchall()

print(f"Printing {len(rows)} rows")
for row in rows:
    print(f"Brand = {row[0]}  Model = {row[1]}  Price = {row[2]}")

cur.close()
conn.close()Code language: PHP (php)

Expected output (actual values will depend on row order):

Printing 2 rows
Brand = volkswagen  Model = golf   Price = 480
Brand = audi        Model = None   Price = 18300

You can also connect pgAdmin to the running container and browse the cars_table visually at http://localhost:5050.

Key Takeaways

• SparkSession is always your starting point in PySpark; local[*] mode lets you develop and test on a laptop before moving to a cluster.
• Define schemas explicitly in production ETL jobs to avoid slow inference scans and type ambiguity.
• Docker Compose eliminates the need to install database engines locally—a single YAML file gives you a reproducible environment for development and testing.
• Psycopg2 is straightforward for loading moderate-sized DataFrames into PostgreSQL; for very large datasets consider using the Spark JDBC writer instead of collecting data to the driver.
• Always call conn.commit() after DML statements when using Psycopg2—auto-commit is off by default.

Next Steps

This local setup is a solid starting point. From here you can:

• Replace collect() + Psycopg2 with Spark’s native JDBC writer for better throughput on large datasets.
• Containerise the PySpark job itself so the entire stack—Spark, Postgres, and pgAdmin—runs inside Docker Compose.
• Add data-quality checks (e.g. row counts, null-rate assertions) between the Transform and Load stages.
• Schedule the job with Apache Airflow or a cron task to run on a cadence.