How to Build an ETL Pipeline in Python

Python has ranked as the most used programming language in the Stack Overflow Developer Survey. Python connects to virtually any data source, handles transformation logic in a few lines of code, and scales from a local CSV script to a distributed Spark job without changing the core architecture.

For ETL, Python gives data teams control that managed tools don’t always offer. You can write extraction logic for non-standard APIs, apply custom transformation rules that GUI-based tools can’t express, and plug the entire pipeline into an orchestrator like Apache Airflow or Prefect. The pipeline does exactly what you design it to do.

• Connect to SQL databases, REST APIs, cloud storage, and flat files from a single script
• Transform data using pandas for in-memory workloads or PySpark for large-scale distributed jobs
• Load into any destination: PostgreSQL, Snowflake, BigQuery, Redshift, S3, or local files
• Orchestrate and schedule with Apache Airflow, Prefect, or cron

This guide walks through how to build an ETL pipeline in Python step by step, with working code examples. We cover the libraries you need at each phase, real-world use cases, best practices for production, and when a managed ETL platform makes more sense than custom Python.

Why Should You Use Python for ETL Processes?

Python has become a go-to choice for data engineers because of its broad range of functionalities. Let’s explore how it is benefiting organizations:

1. Vast ecosystem

As discussed earlier, Python provides extensive libraries to expedite ETL pipelines. Such as:

• Numpy and pandas for data manipulation.
• SQLAlchemy for sqlite3 for database connectivity.
• requests for API interaction.

Due to its vast ecosystem, data engineers are able to simplify complex data aggregation and integration tasks, boosting pipeline efficiency.

2. Readability

Python syntax is designed to be intuitive and close to natural language, which makes the code easy to write, read, and maintain even for new users. This reduces the barrier to entry:

• The easy syntax helps analysts quickly identify logic errors and modify pipelines.
• The syntax simplicity fosters effective collaboration on major ETL projects.
• Emphasis on clean, readable code simplifies maintenance with evolving technical requirements.

The end benefit is that enhanced readability enables quick prototyping and iterative development for faster optimization of ETL logic.

3. Customization

Unlike traditional ETL, Python-driven ETL workflows can be modularized to meet specific business requirements. Given its high customizability, Python scripts can be implemented for:

• Non-standard data sources
• Complex business logic
• Intricate data relationships
• Additional processing steps

This enhanced flexibility allows you to add new elements without re-architecturing the entire pipeline.

4. Automation & Orchestration

You can automate Python-based ETL jobs and orchestrate the execution order with scheduling tools.

• Use built-in modules like cron, schedule, and subprocess.
• Trigger event-based workflows based on timestamp and upstream job status.
• Leverage built-in monitoring, logging, and alerting mechanisms.
• Integrate pipelines with orchestrators like Apache Airflow, Prefect, or Luigi.

5. Strong community & documentation

Python’s ETL ecosystem is backed by an active programming community. The key benefits of the community-driven support include:

• Many ETL libraries (like pandas, Airflow, boto3, etc.) are open-source with active contributors and frequent updates.
• Developers contribute via GitHub by adding new features, fixing bugs, improving, and creating integrations.
• Python-based ETL tools provide detailed documentation and user guides for smooth operation.

An added advantage is the multilingual support and networking opportunities via chat forums and online communities.

How To Build an ETL Pipeline in Python?

Integrating Python with ETL workflows requires a series of technical steps to develop a scalable system. Here’s a step-by-step guide to using a Python script for ETL:

Step 1: Set up your environment

Install Python from python.org, then install the libraries you’ll need:

bash

pip install pandas sqlalchemy

Organize your project directory before writing any code:

etl_pipeline/

├── dealership_data/        # Raw source files (CSV, JSON, XML)

├── etl_pipeline.py         # Main ETL script

├── etl_log.txt             # Pipeline execution log

└── dealership_output.csv   # Transformed output

Configure your file paths at the top of the script:

python

import glob

import pandas as pd

import xml.etree.ElementTree as ET

from datetime import datetime

DATA_DIR = “dealership_data”

TARGET_FILE = “dealership_output.csv”

LOG_FILE = “etl_log.txt”

Step 2: Plan your pipeline

Before writing any functions, define three things:

• Source: Where is the data coming from? (CSV, JSON, and XML files in dealership_data/)
• Transformations: What needs to change? (Remove nulls, deduplicate, standardize price format)
• Destination: Where does it land? (A consolidated CSV for local use, or a database for production)

This mapping prevents scope creep and makes failures easier to diagnose.

Step 3: Extract data from sources

Write a separate extraction function for each file type so each can be tested on its own.

For CSV:

python

def extract_from_csv(path):

    return pd.read_csv(path)

For JSON:

python

def extract_from_json(path):

    return pd.read_json(path, lines=True)

For XML:

python

def extract_from_xml(path):

    cols = [‘car_model’, ‘year_of_manufacture’, ‘price’, ‘fuel’]

    rows = []

    tree = ET.parse(path)

    for elem in tree.getroot():

        rows.append({

            ‘car_model’: elem.find(‘car_model’).text,

            ‘year_of_manufacture’: int(elem.find(‘year_of_manufacture’).text),

            ‘price’: float(elem.find(‘price’).text),

            ‘fuel’: elem.find(‘fuel’).text

        })

    return pd.DataFrame(rows, columns=cols)

Combine all three into a single extract function that loops through the directory:

python

def extract(data_dir):

    df = pd.DataFrame(columns=[‘car_model’, ‘year_of_manufacture’, ‘price’, ‘fuel’])

    extractors = [

        (“*.csv”, extract_from_csv),

        (“*.json”, extract_from_json),

        (“*.xml”, extract_from_xml)

    ]

    for pattern, func in extractors:

        for filepath in glob.glob(f”{data_dir}/{pattern}”):

            df = pd.concat([df, func(filepath)], ignore_index=True)

    return df

glob.glob() finds all matching files in the directory. Each gets passed to the right extractor and the results are combined into one DataFrame.

Step 4: Transform the data

Clean the raw data before it moves to the destination. This step removes nulls, deduplicates records, and standardizes column formats:

python

def transform(df):

    # Drop rows missing critical fields

    df = df.dropna(subset=[‘car_model’, ‘price’])

    # Remove duplicate records

    df = df.drop_duplicates()

    # Round price to two decimal places

    df[‘price’] = df[‘price’].round(2)

    # Coerce year to numeric, set invalid values to NaN

    df[‘year_of_manufacture’] = pd.to_numeric(

        df[‘year_of_manufacture’], errors=’coerce’

    )

    return df

Keep all transformation logic inside this function. If a business rule changes, you update it in one place.

Step 5: Load into the destination

For a local output, write the cleaned DataFrame to a CSV:

python

def load(df, target_file):

    df.to_csv(target_file, index=False)

    print(f”Loaded {len(df)} rows to {target_file}”)

For a production pipeline loading into a SQL database, swap in SQLAlchemy:

python

from sqlalchemy import create_engine

def load_to_db(df, connection_string, table_name):

    engine = create_engine(connection_string)

    df.to_sql(table_name, con=engine, if_exists=’replace’, index=False)

Use if_exists=’append’ to add rows without overwriting, or if_exists=’replace’ to reload the table each run. For cloud warehouses like Snowflake or BigQuery, use their native Python connectors (snowflake-connector-python, google-cloud-bigquery) instead of SQLAlchemy.

Step 6: Add logging

Log each stage with a timestamp so you can track how long each phase takes and pinpoint where failures occur:

python

def log_event(message):

    timestamp = datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’)

    entry = f”{timestamp} – {message}\n”

    with open(LOG_FILE, “a”) as f:

        f.write(entry)

    print(entry.strip())

Call log_event() at the start and end of each phase.

Step 7: Orchestrate the pipeline

Wire the functions together in a single run function. Wrap everything in a try/except so the pipeline catches failures without crashing silently:

python

def run_pipeline():

    log_event(“ETL job started”)

    try:

        # Extract

        log_event(“Extract phase started”)

        raw_data = extract(DATA_DIR)

        log_event(f”Extract complete – {len(raw_data)} rows loaded”)

        # Transform

        log_event(“Transform phase started”)

        clean_data = transform(raw_data)

        log_event(f”Transform complete – {len(clean_data)} rows after cleaning”)

        # Load

        log_event(“Load phase started”)

        load(clean_data, TARGET_FILE)

        log_event(“Load phase complete”)

    except Exception as e:

        log_event(f”Pipeline failed: {e}”)

        raise

    log_event(“ETL job complete”)

if __name__ == “__main__”:

    run_pipeline()

For local automation, schedule this with cron on Linux/macOS or Task Scheduler on Windows.

Scaling to production: Once your pipeline handles real volume, three things matter. First, replace pandas with PySpark for distributed processing across large datasets. Second, switch from CSV to Parquet for the load format: Parquet is columnar and reads significantly faster for analytical queries. Third, move from cron scripts to Apache Airflow or Dagster, which let you define this pipeline as a DAG, schedule runs, track failures, and retry individual stages automatically. For data validation before each load, Great Expectations integrates cleanly with Python pipelines and catches schema mismatches or missing values before they reach your destination.

What Is An Example of Python ETL?

Now we’ll understand the workflow of a Python ETL script with an example. Let’s take a sample use case:

We have to process a mixed-format dataset containing used car listings (that includes CSV, JSON, and XML), apply transformations to standardize raw datasets, and generate a filtered, consolidated CSV file. We’ll track execution via a log in the process.

Let’s start!

Step 1: Import libraries and configure data paths

We use the following code to import necessary libraries and define life paths for input (raw data), output (filtered CSV), and logging:

import glob
import pandas as pd
import xml.etree.ElementTree as ET
from datetime import datetime

What’s happening:

• glob: Find files with similar patterns
• pandas: For tabular data processing
• xml.etree.ElementTree: Parses XML data
• datetime: Adds timestamps to logs

data_dir = "dealership_data"
target_file = "dealership_transformed_data.csv"
log_file = "dealership_logfile.txt"

Explanation:

• data_dir: Source directory containing raw input files.
• target_file: The path where transformed data is written.
• log_file: Captures ETL activity with timestamps for monitoring.

Step 2: Data extraction

Here we define an extraction function for each data type: CSV, JSON, and XML, to retrieve consistent data formats for downstream processing.

For CSV:

def extract_from_csv(path):
return pd.read_csv(path)

• Read the CSV file using pandas.
• Returns a DataFrame for downstream processing.

For JSON:

def extract_from_json(path):
return pd.read_json(path, lines=True)

• Read the JSON file using pandas.
• Used when each subject of the JSON file is stored on a separate line (lines=true).

For XML,

def extract_from_xml(path):
cols = ['car_model','year_of_manufacture','price','fuel']
rows = []
tree = ET.parse(path)
for elem in tree.getroot():
rows.append({
'car_model': elem.find("car_model").text,
'year_of_manufacture': int(elem.find("year_of_manufacture").text),'
price': float(elem.find("price").text),
'fuel': elem.find("fuel").text
})
return pd.DataFrame(rows, columns=cols)

• Parses the XML file, navigates its structure, and extracts specific fields.
• Constructs a list of dictionaries and converts it to a DataFrame.

Here, the consistent column names are car_model, year_of_manufacture, price, and fuel.

Step 3: Combine data from all files

The following code loops through the files in the dealership data factory, applies the accurate extractor, and creates a unified DataFrame with all the raw data.

def extract_all(data_dir):
df = pd.DataFrame(columns=['car_model','year_of_manufacture','price','fuel'])
for pattern, func in [
("*.csv", extract_from_csv),
("*.json", extract_from_json),
("*.xml", extract_from_xml)
]:
for filepath in glob.glob(f"{data_dir}/{pattern}"):
df = pd.concat([df, func(filepath)], ignore_index=True)
return df

What’s happening:

• glob is used to find matching files in the directory.
• Iterates through each file type.

Step 4: Data transformation

Here we clean and prepare the dataset by removing null values, deduplicating to avoid redundancy, and formatting numerical columns.

def transform(df):
df = df.dropna(subset=['price', 'car_model'])
df = df.drop_duplicates()
df['price'] = df['price'].round(2)
df['year_of_manufacture'] = pd.to_numeric(df['year_of_manufacture'], errors='coerce')
return df

Explanation:

• Remove nulls in the price and car_model columns.
• Converts years_of_manufacture to numeric values.
• Rounds [df['price'].round(2)] price value to two decimals.

Step 5: Data loading and log events

Now, we’ll load the data and implement log events.

For loading,

def load_to_csv(df, target_file):<br>df.to_csv(target_file, index=False)

What’s happening:

• load_to_csv saves the transformed DataFrame to a CSV file.
• index=False prevents row indices from being written as an extra column.

For logging,

def log_event(message):<br>ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')<br>with open(log_file, "a") as f:<br>f.write(f"{ts} - {message}\n")

Explanation:

• log_event timestamps for transparency and debugging.
• Identifies failure points, monitors each stage, and time taken for execution.

Step 6: Orchestration of the ETL pipeline

Now we run the ETL process step-by-step:

log_event("ETL Job Started")

This logs the beginning of the ETL process.

For extraction,

log_event("Extract phase Started")
df = extract_all(data_dir)
log_event("Extract phase Ended")

• Logs the start of the extraction process.
• extract_all(data_dir) reads the files and merges the results in a single DataFrame.

For transformation,

log_event("Transform phase Started")<br>df = transform(df)<br>log_event("Transform phase Ended")

• Drops rows with missing prices or car models.
• Removes duplicates and ensures correct representation of year_of_manufacture.

For loading,

log_event("Load phase Started")
load_to_csv(df, target_file)
log_event("Load phase Ended")

• load_to_csv() writes the cleansed dataset to the target file.
• dealership_transformed_data.csv is the target file that stores datasets for analytics and reporting.

To end the ETL job,

log_event("ETL Job Ended")

The closing log ensures pipeline execution without any interruption.

Real-World Use Cases of Python ETL

Here are some detailed use cases of Python ETL:

1. Real-time data streaming

Use case: Logistics companies require real-time shipment status to optimize routes and provide delivery updates.

Python scripts process live data streams using libraries like kafka-python for stream ingestion. Python ETL scripts extract data from APIs and IoT devices, enrich it with delivery routes, and load it into a database for real-time analytics. 

The ability to manage streaming data and integrate with distributed systems enables instant notification and route optimization.

2. Cross-language interoperability

Use case: Companies integrate data from systems written in distinct languages (Java, C++) and diverse platforms (Windows, macOS).

You can leverage its interoperability layer to connect with:

• Java via JPype or Py4J
• C/C++ using ctypes, cffi, or SWIG
• R via rpy2
• .NET applications through pythonnet

This makes Python a “glue language” and acts as a universal orchestrator in cross-platform environments. Companies can execute external programs, integrate with native modules, and operate across OS boundaries to span legacy systems and modern cloud services within a unified Python pipeline.

3. Lightweight and scriptable ETL

Use case: Teams quickly ingest data from new data sources without setting up a full ETL infrastructure.

Data engineers can craft quick scripts using pandas to read CSV and JSON files, filter datasets, and load them into a destination in a few lines of code. These scripts are version-controlled and are ideal for ad hoc tasks, proofs-of-concept, or environments where agility is more important than orchestration. 

This highlights rapid scripting capabilities appropriate for isolated tasks.

4. Customer 360-degree view

Use case: Brands want a comprehensive view of customer behavior by integrating with CRMs, transactional databases, and analytical platforms.

As discussed earlier, Python scripts extract data from various sources, including:

• APIs like Salesforce and HubSpot use requests or platform SDKs.
• Campaign data from Mailchimp and Google Ads.
• Transaction records from SQL databases through sqlite3 or psycopg2.

The gathered data is normalized into unified datetime formats and deduplicated based on customer ID or email, and transformed into a consolidated DataFrame. Lastly, it is loaded into a data warehouse or exported to BI tools for dashboards and visual reports.

5. Automated data analysis

Use case: Healthcare institutions have to generate automated patient reports for operational efficiency.

Python scripts are scheduled with cron and orchestrators like Airflow or Luigi to pull patient data from electronic health record (EHR) via SQL connectors (psycopg2 or pyodbc) and pandas. 

The raw data is standardized and enriched with derived metrics like admission rates and occupied beds. Finally, the processed data is exported as a PDF or Excel for report distribution.

6. E-commerce inventory management

Use case: Retailers synchronize inventory data across multiple streams to maintain accurate stock levels.

The data extracted from supplier APIs and internal databases is transformed to reconcile discrepancies, standardize units, and calculate inventory in real-time. The processed data is transferred to a centralized repository like PostgreSQL to drive stock alerts and update dashboards with real-time data feeds.

Best Practices When Building an ETL Pipeline in Python

The best practices to build a resilient and scalable ETL pipeline in Python are:

1. Modular design

Focus on breaking down your pipeline into smaller modules to handle specific tasks. This enhances code reusability, expedites debugging, and simplifies future modifications in the pipeline.

• Isolate extract(), transform(), and load() to develop, test, and maintain separately.
• Design modules to reuse across various ETL projects and pipelines.
• Use configuration files like YAML and JSON to manage different data environments with minimal coding.

A modular design enables teams to maintain individual components and independent testing of modules through unit testing and mocking.

2. Fault tolerance

Deploy comprehensive error-handling mechanisms at every stage of the pipeline to promote self-recovery and pipeline continuity. You can use built-in logging modules to detect log errors and dependencies.

Ensure that re-executing the ETL script doesn’t result in duplicate records and data inconsistencies. For fault tolerance:

• Use structured logging and context-rich error messages to facilitate debugging.
• Leverage orchestration frameworks to apply built-in retry logic and failure handling.
• Isolate rows that fail validation by writing them to a separate table for later inspection.

3. Pipeline testing

Testing is necessary to validate the dataset and ensure ETL pipeline accuracy.

• Run unit tests and use Python testing frameworks like pytest or unittest to validate outputs, spot regressions when logic changes, and mock external dependencies.
• Run integration tests to validate end-to-end execution, data flow between modules, and compatibility between sources and destinations.
• Deploy schema and quality checks to validate dataframes.

4. Scalability

Prepare the pipeline to work with vast datasets for effective scalability. Implement features like:

• Incremental loading with CDC to avoid full reloads.
• Parallel processing with Python libraries like concurrent.futures, multiprocessing, or frameworks like dask for handling large datasets.
• Distributed execution using Spark to deal with high-throughput and multi-source data pipelines.
• Adopt version control (Git) and CI/CD practices to monitor changes and automate deployment.

5. Big data integration

Integrate with big data platforms and services to handle large datasets.

• Cloud SDKs: Python libraries offer boto3 to connect with cloud-based data lakes and storage services. These SDKs enable secure, scalable, and code-driven data ingestion and distribution pipelines.
• Serverless platforms: Deploy Python ETL scripts on serverless platforms like Google Cloud Functions, AWS Lambda, and Azure Data Factory to process large volumes with minimum management.

Moreover, as Python supports various big data formats, leverage its flexibility in data ingestion while ensuring compatibility with diverse data ecosystems.

Python vs Other Programming Languages for ETL

Python is the default choice for most ETL work, but other languages have legitimate use cases depending on your team’s background, data scale, and infrastructure. Here’s how the main alternatives compare:

Language	Best for	Key ETL libraries	Where it beats Python
Python	End-to-end ETL from prototypes to production	pandas, PySpark, SQLAlchemy, Airflow, boto3	Baseline: widest library support, fastest to build, most community resources
Java	Enterprise-grade, high-throughput pipelines in JVM environments	Apache Spark, Hadoop, Apache Kafka, Spring Batch	Stricter type safety, better thread management, native integration with Hadoop/Spark at scale
Scala	Distributed big data ETL with Spark	Apache Spark (native), Kafka Streams, Akka	Native Spark language; more performant and expressive than PySpark for complex Spark jobs
Go	High-performance, low-latency ETL scripts	Apache Beam (Go SDK), custom connectors	Faster execution, lower memory footprint, better for high-throughput streaming pipelines
Ruby	Lightweight scripting, niche transformation tasks	Kiba, Nokogiri, Square’s ETL gem	Simpler DSL for transformation logic; cleaner syntax for small, focused scripts

For most teams starting or running mid-scale pipelines, Python is the right call. Scala becomes worth considering when Spark performance is a bottleneck. Java and Go are usually chosen based on existing team expertise, not ETL-specific advantages.

Conclusion

This article provided information on Python, its key features, different methods to set up ETL using Python scripts, limitations of manually setting up ETL using Python, top Python libraries to set up an ETL pipeline, and top ETL tools using Python. To take your automation further, see how teams use Python for Data Engineering to orchestrate pipelines and manage production workloads efficiently.

If you’re looking for a no-code solution to simplify your ETL process, Hevo is an excellent choice. Sign up for a 14-day free trial and simplify your data integration process. Check out the pricing details to understand which plan fulfills all your business needs.

Frequently Asked Questions