# Why Breaking Things Apart Eventually Brings Them Together: Modulazition in ETL

What does this big word really mean?

The first week of data engineering at luxdev was interesting, especially engaging was the ETL(extraction, transformation and loading) Workflows that had me on edge. At the end of the week we had to finish a project- an end to end pipeline that extracted data from an API, modified a few columns and eventually loaded it to an sql database. All this processes were easily streamlined by a proper knowledge of modularization.

Modularization in ETL is exactly what it sounds like,breaking your data pipeline into smaller, independent, self-contained pieces. Each piece does one thing. It does that thing well. And it can be swapped, tested, or understood without reading the entire codebase.

Pipeline.py

Check out the pipeline below. Isnt this a mouthful really?

import http.client
import json
import pandas as pd
import os
from sqlalchemy import create_engine, text
from dotenv import load_dotenv

load_dotenv()

#data extraction

def extract_prices():
    conn = http.client.HTTPSConnection("api.collectapi.com")
    headers = {
        'content-type': "application/json",
        'authorization': "apikey 5hqygBScH1ifb1p6OdZodL:6uYLDoWQxkYaI5r4oHoxvr"
        }
    conn.request("GET", "/gasPrice/stateUsaPrice?state=WA", headers=headers)
    res = conn.getresponse()
    data = res.read()
    data_string = data.decode("utf-8")
    data_list = json.loads(data_string)
    return data_list


#data transformation

def transform_prices(data_list):
    city_level = data_list['result']['cities']
    city_df = pd.DataFrame(city_level)
    city_df = city_df.drop(columns=['lowername'])
    city_df = city_df.rename(columns={'name': 'cities'})
    return city_df

#data loading

def load_prices(city_df):
    DATABASE_NAME = os.getenv('DATABASE_NAME')
    DATABASE_USER = os.getenv('DATABASE_USER')
    DATABASE_PORT = os.getenv('DATABASE_PORT')
    DATABASE_PASSWORD = os.getenv('DATABASE_PASSWORD')
    DATABASE_HOST = os.getenv('DATABASE_HOST')
    engine = create_engine(f'postgresql://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}')
    with engine.connect() as conn:
        resort = conn.execute(text('select * from cities;'))
        for i in resort:
            print(i)
    city_df.to_sql('cities', engine, if_exists='replace', index=False)


def main():
    data_list = extract_prices()
    city_df = transform_prices(data_list)
    load_prices(city_df)
    print('ETL process completed successfully.')

    if __name__ == "__main__":
        main()

It works. But it's a single file with everything tangled together. The extraction logic sits right next to the database credentials. The transformation knows nothing about where data comes from or where it goes. If the API changes, you're scrolling past the database code to find the HTTP call. If the database schema changes, you're hunting through the API logic.

Now here's the same pipeline, broken into pieces.

The Extraction Phase:

The extraction layer is where modularity earns its keep. Here's what a focused extract module looks like:

# extract.py
import http.client
import json

def extract_prices():
    conn = http.client.HTTPSConnection("api.collectapi.com")
    headers = {
        'content-type': "application/json",
        'authorization': "apikey 5hqygBScH1ifb1p6OdZodL:6uYLDoWQxkYaI5r4oHoxvr"
        }
    conn.request("GET", "/gasPrice/stateUsaPrice?state=WA", headers=headers)
    res = conn.getresponse()
    data = res.read()
    data_string = data.decode("utf-8")
    data_list = json.loads(data_string)
    return data_list

Thirty lines. One job. It talks to the API, parses the JSON, and hands it off. It doesn't know about DataFrames, database connections, or environment variables. It doesn't care. When the API changes its endpoint or adds authentication, you open one file and fix it. Done.

The Transform Phase:

This is where changes are made, to columns, rename this , drop that.

Can you spot what is going on below:

# transform.py
import pandas as pd

def transform_prices(data_list):
    city_level = data_list['result']['cities']
    city_df = pd.DataFrame(city_level)
    city_df = city_df.drop(columns=['lowername'])
    city_df = city_df.rename(columns={'name': 'cities'})
    return city_df

Short. Sweet. Single responsibility. It takes the raw JSON dict, builds a DataFrame, drops the column nobody needs, and renames one field to be clearer. You can test this function in isolation. You can understand it in ten seconds. You can change it without worrying about breaking the API call or the database write.

The beautiful thing about modular transformations is that they become testable. You can write a unit test for your column rename without setting up a database connection. You can validate the DataFrame shape without running the entire pipeline. When something breaks, and something always breaks, you isolate the failure to a single module.

The Load Phase:

Loading usually gets less attention in modularity discussions because, frankly, it's the easy part. But even here, modular thinking pays off.

# load.py
import os
from sqlalchemy import create_engine, text
from dotenv import load_dotenv

load_dotenv()

def load_prices(city_df):
    DATABASE_NAME = os.getenv('DATABASE_NAME')
    DATABASE_USER = os.getenv('DATABASE_USER')
    DATABASE_PORT = os.getenv('DATABASE_PORT')
    DATABASE_PASSWORD = os.getenv('DATABASE_PASSWORD')
    DATABASE_HOST = os.getenv('DATABASE_HOST')

    engine = create_engine(
        f'postgresql://{DATABASE_USER}:{DATABASE_PASSWORD}@'
        f'{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}'
    )

    with engine.connect() as conn:
        resort = conn.execute(text('select * from cities;'))
        for i in resort:
            print(i)

    city_df.to_sql('cities', engine, if_exists='replace', index=False)

Maybe today you're loading into PostgreSQL. Next quarter, the C-suite might decide everything needs to live in Snowflake. If your load logic is tangled up with your transformation logic, you're looking at a rewrite. If it's its own module, you write a new destination adapter and plug it in.

Your extraction doesn't care where the data ends up. Your transformations don't either. The load module is just another box in the toolbox.

Tying It All Together

The beauty of a modular pipeline is how little the orchestrator needs to do:

# main.py
from extract import extract_prices
from transform import transform_prices
from load import load_prices

def main():
    data_list = extract_prices()
    city_df = transform_prices(data_list)
    load_prices(city_df)
    print('ETL process completed successfully.')

if __name__ == "__main__":
    main()

That's it. Three imports, three function calls, one print statement. The orchestrator doesn't know how extraction works. It doesn't know how loading works. It just knows the order in which things happen. Each module handles its own complexity behind a clean function signature.

Compare this to the gigantic pipeline.py version above. Which one would you rather inherit?