ETL with Python: Extracting, Transforming, and Loading Data

ETL stands for Extract, Transform, and Load. It is a data-integration process that collects data from one or more sources, converts it into a consistent and usable structure, and writes the result to a destination system.

The three stages run in this order:

Extract → Transform → Load

Extract → Transform → Load

For example, an organization might receive height and weight data in CSV and JSON files. The files use imperial units, but the destination application requires metric units and a consistent tabular structure.

An ETL pipeline can:

1. Extract records from every source file.
2. Validate and combine the records.
3. Convert inches to meters and pounds to kilograms.
4. Load the transformed data into a target CSV file.
5. Record the pipeline’s progress and failures in a log.

Is ETL the Same as Batch Processing?

ETL is commonly implemented as batch processing, but the concepts are not identical.

Batch processing handles a collection of records at scheduled or defined intervals. For example, a pipeline might process all files received during the previous day.

ETL describes the extract-transform-load sequence. ETL pipelines can operate in:

• Scheduled batches
• Micro-batches
• Event-driven workflows
• Near-real-time or streaming systems

A simple file-based ETL program is usually a batch pipeline because it reads a group of files and processes them together.

The Three ETL Stages

1. Extract

Extraction retrieves data from source systems without yet applying the complete set of business transformations.

Sources can include:

• CSV files
• JSON files
• Spreadsheets
• Relational databases
• NoSQL databases
• Web APIs
• Message brokers
• Cloud object storage
• Application logs

An extraction process should preserve enough source information to support validation and troubleshooting.

Useful extraction metadata can include:

• Source filename
• Source system
• Extraction timestamp
• Batch identifier
• Original record identifier

2. Transform

Transformation converts extracted data into the structure and quality required by the destination.

Typical transformations include:

• Renaming columns
• Converting data types
• Standardizing units
• Parsing dates
• Removing duplicates
• Handling missing values
• Joining datasets
• Applying business rules
• Validating accepted ranges
• Creating derived columns
• Masking sensitive information

Transformations should be explicit and testable. Silent corrections or discarded records can conceal data-quality problems.

3. Load

Loading writes the transformed data into its destination.

Possible destinations include:

• A CSV or Parquet file
• A relational database
• A data warehouse
• A data lake
• An analytical table
• A downstream API
• A machine-learning feature store

A production load process should consider duplicate execution, partial failure, transactions, retry behavior, and recovery.

Example ETL Scenario

Suppose the input directory contains the following files:

data/input/customers_01.csv
data/input/customers_02.csv
data/input/customers_03.json

data/input/customers_01.csv
data/input/customers_02.csv
data/input/customers_03.json

Every file represents records with this logical schema:

The target dataset should contain:

The conversion formulas are:\[ \text{height in meters} = \text{height in inches}\times 0.0254 \]\[ \text{weight in kilograms} = \text{weight in pounds}\times 0.45359237 \]

The original transcript alternates between describing height as inches and feet. Those units are not interchangeable. This implementation explicitly assumes that the source column contains inches.

Project Structure

A small project can use the following directory structure:

etl-project/
├── data/
│   ├── input/
│   │   ├── customers_01.csv
│   │   ├── customers_02.csv
│   │   └── customers_03.json
│   └── output/
├── logs/
└── etl.py

etl-project/
├── data/
│   ├── input/
│   │   ├── customers_01.csv
│   │   ├── customers_02.csv
│   │   └── customers_03.json
│   └── output/
├── logs/
└── etl.py

Keeping input and output files in separate directories prevents the pipeline from accidentally treating its own output as new source data.

Sample CSV Data

name,height_inches,weight_pounds
Alice,65,130
Bob,70,180
Carla,62,115

name,height_inches,weight_pounds
Alice,65,130
Bob,70,180
Carla,62,115

Sample JSON Data

The following example uses a JSON array of records:

[
  {
    "name": "Daniel",
    "height_inches": 72,
    "weight_pounds": 200
  },
  {
    "name": "Elena",
    "height_inches": 64,
    "weight_pounds": 125
  }
]

[
  {
    "name": "Daniel",
    "height_inches": 72,
    "weight_pounds": 200
  },
  {
    "name": "Elena",
    "height_inches": 64,
    "weight_pounds": 125
  }
]

A line-delimited JSON file is structured differently:

{"name":"Daniel","height_inches":72,"weight_pounds":200}
{"name":"Elena","height_inches":64,"weight_pounds":125}

{"name":"Daniel","height_inches":72,"weight_pounds":200}
{"name":"Elena","height_inches":64,"weight_pounds":125}

For line-delimited JSON, pandas must be called with lines=True. The pipeline must therefore know which JSON representation its sources use. pandas read_json() documentation

Complete Python ETL Program

from __future__ import annotations

import logging
from pathlib import Path

import pandas as pd


INPUT_DIRECTORY = Path("data/input")
OUTPUT_FILE = Path("data/output/people_metric.csv")
LOG_FILE = Path("logs/etl.log")

REQUIRED_COLUMNS = {
    "name",
    "height_inches",
    "weight_pounds",
}


def configure_logging() -> None:
    """Configure logging to both a file and the console."""
    LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        handlers=[
            logging.FileHandler(LOG_FILE, encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )


def read_csv_file(file_path: Path) -> pd.DataFrame:
    """Read one CSV file into a DataFrame."""
    logging.info("Reading CSV file: %s", file_path)
    return pd.read_csv(file_path)


def read_json_file(file_path: Path) -> pd.DataFrame:
    """Read one JSON array-of-records file into a DataFrame."""
    logging.info("Reading JSON file: %s", file_path)
    return pd.read_json(file_path, orient="records")


def validate_schema(
    data: pd.DataFrame,
    source_name: str,
) -> None:
    """Verify that a source contains every required column."""
    missing_columns = REQUIRED_COLUMNS - set(data.columns)

    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(
            f"{source_name} is missing required columns: {missing}"
        )


def extract(input_directory: Path) -> pd.DataFrame:
    """Read and combine supported files from the input directory."""
    if not input_directory.exists():
        raise FileNotFoundError(
            f"Input directory does not exist: {input_directory}"
        )

    source_files = sorted(
        [
            *input_directory.glob("*.csv"),
            *input_directory.glob("*.json"),
        ]
    )

    if not source_files:
        raise FileNotFoundError(
            f"No CSV or JSON files found in {input_directory}"
        )

    extracted_frames: list[pd.DataFrame] = []

    for file_path in source_files:
        if file_path.suffix.lower() == ".csv":
            frame = read_csv_file(file_path)
        elif file_path.suffix.lower() == ".json":
            frame = read_json_file(file_path)
        else:
            continue

        validate_schema(frame, file_path.name)

        frame = frame.copy()
        frame["source_file"] = file_path.name
        extracted_frames.append(frame)

        logging.info(
            "Extracted %d rows from %s",
            len(frame),
            file_path.name,
        )

    extracted_data = pd.concat(
        extracted_frames,
        ignore_index=True,
    )

    logging.info(
        "Extraction complete: %d files and %d rows",
        len(extracted_frames),
        len(extracted_data),
    )

    return extracted_data


def transform(data: pd.DataFrame) -> pd.DataFrame:
    """Validate values and convert imperial units to metric units."""
    transformed = data.copy()

    transformed["height_inches"] = pd.to_numeric(
        transformed["height_inches"],
        errors="coerce",
    )

    transformed["weight_pounds"] = pd.to_numeric(
        transformed["weight_pounds"],
        errors="coerce",
    )

    invalid_mask = (
        transformed["name"].isna()
        | transformed["height_inches"].isna()
        | transformed["weight_pounds"].isna()
        | (transformed["height_inches"] <= 0)
        | (transformed["weight_pounds"] <= 0)
    )

    if invalid_mask.any():
        invalid_rows = transformed.loc[
            invalid_mask,
            [
                "name",
                "height_inches",
                "weight_pounds",
                "source_file",
            ],
        ]

        logging.error(
            "Transformation stopped because %d rows are invalid:\n%s",
            len(invalid_rows),
            invalid_rows.to_string(index=False),
        )

        raise ValueError(
            f"Found {len(invalid_rows)} records with missing, "
            "non-numeric, or non-positive required values"
        )

    transformed["height_meters"] = (
        transformed["height_inches"] * 0.0254
    ).round(2)

    transformed["weight_kilograms"] = (
        transformed["weight_pounds"] * 0.45359237
    ).round(2)

    transformed = transformed[
        [
            "name",
            "height_meters",
            "weight_kilograms",
            "source_file",
        ]
    ]

    logging.info(
        "Transformation complete: %d rows",
        len(transformed),
    )

    return transformed


def load(data: pd.DataFrame, output_file: Path) -> None:
    """Write transformed data to the target CSV file."""
    output_file.parent.mkdir(parents=True, exist_ok=True)

    temporary_file = output_file.with_suffix(".tmp")

    data.to_csv(
        temporary_file,
        index=False,
        encoding="utf-8",
    )

    temporary_file.replace(output_file)

    logging.info(
        "Loaded %d rows into %s",
        len(data),
        output_file,
    )


def run_pipeline() -> None:
    """Execute the ETL stages in the required order."""
    logging.info("ETL pipeline started")

    extracted_data = extract(INPUT_DIRECTORY)
    transformed_data = transform(extracted_data)
    load(transformed_data, OUTPUT_FILE)

    logging.info("ETL pipeline completed successfully")


def main() -> None:
    configure_logging()

    try:
        run_pipeline()
    except Exception:
        logging.exception("ETL pipeline failed")
        raise


if __name__ == "__main__":
    main()

from __future__ import annotations

import logging
from pathlib import Path

import pandas as pd


INPUT_DIRECTORY = Path("data/input")
OUTPUT_FILE = Path("data/output/people_metric.csv")
LOG_FILE = Path("logs/etl.log")

REQUIRED_COLUMNS = {
    "name",
    "height_inches",
    "weight_pounds",
}


def configure_logging() -> None:
    """Configure logging to both a file and the console."""
    LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        handlers=[
            logging.FileHandler(LOG_FILE, encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )


def read_csv_file(file_path: Path) -> pd.DataFrame:
    """Read one CSV file into a DataFrame."""
    logging.info("Reading CSV file: %s", file_path)
    return pd.read_csv(file_path)


def read_json_file(file_path: Path) -> pd.DataFrame:
    """Read one JSON array-of-records file into a DataFrame."""
    logging.info("Reading JSON file: %s", file_path)
    return pd.read_json(file_path, orient="records")


def validate_schema(
    data: pd.DataFrame,
    source_name: str,
) -> None:
    """Verify that a source contains every required column."""
    missing_columns = REQUIRED_COLUMNS - set(data.columns)

    if missing_columns:
        missing = ", ".join(sorted(missing_columns))
        raise ValueError(
            f"{source_name} is missing required columns: {missing}"
        )


def extract(input_directory: Path) -> pd.DataFrame:
    """Read and combine supported files from the input directory."""
    if not input_directory.exists():
        raise FileNotFoundError(
            f"Input directory does not exist: {input_directory}"
        )

    source_files = sorted(
        [
            *input_directory.glob("*.csv"),
            *input_directory.glob("*.json"),
        ]
    )

    if not source_files:
        raise FileNotFoundError(
            f"No CSV or JSON files found in {input_directory}"
        )

    extracted_frames: list[pd.DataFrame] = []

    for file_path in source_files:
        if file_path.suffix.lower() == ".csv":
            frame = read_csv_file(file_path)
        elif file_path.suffix.lower() == ".json":
            frame = read_json_file(file_path)
        else:
            continue

        validate_schema(frame, file_path.name)

        frame = frame.copy()
        frame["source_file"] = file_path.name
        extracted_frames.append(frame)

        logging.info(
            "Extracted %d rows from %s",
            len(frame),
            file_path.name,
        )

    extracted_data = pd.concat(
        extracted_frames,
        ignore_index=True,
    )

    logging.info(
        "Extraction complete: %d files and %d rows",
        len(extracted_frames),
        len(extracted_data),
    )

    return extracted_data


def transform(data: pd.DataFrame) -> pd.DataFrame:
    """Validate values and convert imperial units to metric units."""
    transformed = data.copy()

    transformed["height_inches"] = pd.to_numeric(
        transformed["height_inches"],
        errors="coerce",
    )

    transformed["weight_pounds"] = pd.to_numeric(
        transformed["weight_pounds"],
        errors="coerce",
    )

    invalid_mask = (
        transformed["name"].isna()
        | transformed["height_inches"].isna()
        | transformed["weight_pounds"].isna()
        | (transformed["height_inches"] <= 0)
        | (transformed["weight_pounds"] <= 0)
    )

    if invalid_mask.any():
        invalid_rows = transformed.loc[
            invalid_mask,
            [
                "name",
                "height_inches",
                "weight_pounds",
                "source_file",
            ],
        ]

        logging.error(
            "Transformation stopped because %d rows are invalid:\n%s",
            len(invalid_rows),
            invalid_rows.to_string(index=False),
        )

        raise ValueError(
            f"Found {len(invalid_rows)} records with missing, "
            "non-numeric, or non-positive required values"
        )

    transformed["height_meters"] = (
        transformed["height_inches"] * 0.0254
    ).round(2)

    transformed["weight_kilograms"] = (
        transformed["weight_pounds"] * 0.45359237
    ).round(2)

    transformed = transformed[
        [
            "name",
            "height_meters",
            "weight_kilograms",
            "source_file",
        ]
    ]

    logging.info(
        "Transformation complete: %d rows",
        len(transformed),
    )

    return transformed


def load(data: pd.DataFrame, output_file: Path) -> None:
    """Write transformed data to the target CSV file."""
    output_file.parent.mkdir(parents=True, exist_ok=True)

    temporary_file = output_file.with_suffix(".tmp")

    data.to_csv(
        temporary_file,
        index=False,
        encoding="utf-8",
    )

    temporary_file.replace(output_file)

    logging.info(
        "Loaded %d rows into %s",
        len(data),
        output_file,
    )


def run_pipeline() -> None:
    """Execute the ETL stages in the required order."""
    logging.info("ETL pipeline started")

    extracted_data = extract(INPUT_DIRECTORY)
    transformed_data = transform(extracted_data)
    load(transformed_data, OUTPUT_FILE)

    logging.info("ETL pipeline completed successfully")


def main() -> None:
    configure_logging()

    try:
        run_pipeline()
    except Exception:
        logging.exception("ETL pipeline failed")
        raise


if __name__ == "__main__":
    main()

Understanding the Extraction Stage

Discovering Files

Python’s pathlib module can locate files with a particular extension:

csv_files = list(INPUT_DIRECTORY.glob("*.csv"))
json_files = list(INPUT_DIRECTORY.glob("*.json"))

csv_files = list(INPUT_DIRECTORY.glob("*.csv"))
json_files = list(INPUT_DIRECTORY.glob("*.json"))

The asterisk is a wildcard. The pattern *.csv matches filenames ending in .csv.

The program combines and sorts both lists:

source_files = sorted(
    [
        *input_directory.glob("*.csv"),
        *input_directory.glob("*.json"),
    ]
)

source_files = sorted(
    [
        *input_directory.glob("*.csv"),
        *input_directory.glob("*.json"),
    ]
)

Sorting creates a deterministic processing order. This is helpful for testing and troubleshooting, although a production pipeline should not depend on filename order unless that dependency is explicitly designed.

Reading CSV Files

A CSV file can be read into a pandas DataFrame:

frame = pd.read_csv(file_path)

frame = pd.read_csv(file_path)

A DataFrame represents tabular data using labeled rows and columns.

Reading JSON Files

For a JSON array of records:

frame = pd.read_json(
    file_path,
    orient="records",
)

frame = pd.read_json(
    file_path,
    orient="records",
)

For line-delimited JSON, use:

frame = pd.read_json(
    file_path,
    lines=True,
)

frame = pd.read_json(
    file_path,
    lines=True,
)

The format should be explicitly documented rather than determined by repeatedly trying unrelated parsing modes.

Validating the Schema

Before combining files, the pipeline checks whether each source provides the expected columns:

missing_columns = REQUIRED_COLUMNS - set(data.columns)

missing_columns = REQUIRED_COLUMNS - set(data.columns)

If a required column is missing, the pipeline stops with an informative error.

Without this check, a misspelled column such as height_inche could produce missing values later in the pipeline and make the underlying source problem harder to identify.

Recording Source Provenance

The pipeline adds the filename to every extracted record:

frame["source_file"] = file_path.name

frame["source_file"] = file_path.name

If an invalid record is detected later, the filename makes it easier to locate the source.

Combining DataFrames

The extracted frames are collected in a list and combined once:

extracted_data = pd.concat(
    extracted_frames,
    ignore_index=True,
)

extracted_data = pd.concat(
    extracted_frames,
    ignore_index=True,
)

ignore_index=True creates a new sequential index for the combined result:

0, 1, 2, 3, ...

0, 1, 2, 3, ...

Without it, each source’s original index may be preserved, resulting in repeated index values.

The pipeline uses pd.concat() rather than repeatedly appending rows to a DataFrame. Current pandas documentation recommends collecting frames and concatenating them together because iterative concatenation creates unnecessary copies. pandas concatenation documentation

The former DataFrame.append() method should not be used in current pandas code.

Understanding the Transformation Stage

Converting Values to Numeric Types

Values read from external sources may contain strings, blanks, or invalid symbols.

The program explicitly converts measurement columns:

transformed["height_inches"] = pd.to_numeric(
    transformed["height_inches"],
    errors="coerce",
)

transformed["height_inches"] = pd.to_numeric(
    transformed["height_inches"],
    errors="coerce",
)

With errors="coerce", invalid values become missing values. The pipeline then detects them during validation.

This is safer than allowing an invalid string to cause an unexpected calculation later.

Validating Values

The program rejects records where:

• The name is missing.
• Height is missing or non-numeric.
• Weight is missing or non-numeric.
• Height is zero or negative.
• Weight is zero or negative.

invalid_mask = (
    transformed["name"].isna()
    | transformed["height_inches"].isna()
    | transformed["weight_pounds"].isna()
    | (transformed["height_inches"] <= 0)
    | (transformed["weight_pounds"] <= 0)
)

invalid_mask = (
    transformed["name"].isna()
    | transformed["height_inches"].isna()
    | transformed["weight_pounds"].isna()
    | (transformed["height_inches"] <= 0)
    | (transformed["weight_pounds"] <= 0)
)

A production pipeline might quarantine invalid records instead of stopping the complete batch. The appropriate behavior depends on the business requirements.

Possible policies include:

• Reject the complete batch.
• Skip invalid records and issue a warning.
• Write invalid records to a quarantine table.
• Correct values using an approved reference source.
• Continue only when the invalid-record rate remains below a threshold.

The policy should be deliberate and auditable.

Converting Height

One inch is exactly 0.0254 meters:

transformed["height_meters"] = (
    transformed["height_inches"] * 0.0254
).round(2)

transformed["height_meters"] = (
    transformed["height_inches"] * 0.0254
).round(2)

For example:

65 inches × 0.0254 = 1.651 meters

65 inches × 0.0254 = 1.651 meters

Rounded to two decimal places:

1.65 meters

1.65 meters

If the source were actually measured in feet, the correct calculation would be different:

height_meters = height_feet * 0.3048

height_meters = height_feet * 0.3048

The pipeline must never guess the unit from the values alone.

Converting Weight

One pound is approximately 0.45359237 kilograms:

transformed["weight_kilograms"] = (
    transformed["weight_pounds"] * 0.45359237
).round(2)

transformed["weight_kilograms"] = (
    transformed["weight_pounds"] * 0.45359237
).round(2)

For example:

130 pounds × 0.45359237 = 58.967...

130 pounds × 0.45359237 = 58.967...

Rounded to two decimal places:

58.97 kilograms

58.97 kilograms

Selecting the Output Schema

The pipeline retains only the required destination columns:

transformed = transformed[
    [
        "name",
        "height_meters",
        "weight_kilograms",
        "source_file",
    ]
]

transformed = transformed[
    [
        "name",
        "height_meters",
        "weight_kilograms",
        "source_file",
    ]
]

This creates an explicit output contract. Extra source columns do not silently enter the target dataset.

Understanding the Load Stage

The simplest load operation writes the transformed DataFrame to a CSV file:

data.to_csv(
    output_file,
    index=False,
    encoding="utf-8",
)

data.to_csv(
    output_file,
    index=False,
    encoding="utf-8",
)

index=False prevents the pandas row index from being written as an additional column.

Reducing the Risk of Partial Output

Writing directly to the final target can leave a partially written file if the process is interrupted.

The example first writes to a temporary file:

temporary_file = output_file.with_suffix(".tmp")

temporary_file = output_file.with_suffix(".tmp")

It then replaces the target after writing succeeds:

temporary_file.replace(output_file)

temporary_file.replace(output_file)

Renaming or replacing files is generally safer than exposing a partially written result. Exact atomicity guarantees depend on the operating system and filesystem, so production designs must evaluate their storage environment.

Overwrite Versus Append

The example replaces the complete target file on every successful run.

Another pipeline might append new rows:

data.to_csv(
    output_file,
    mode="a",
    header=not output_file.exists(),
    index=False,
)

data.to_csv(
    output_file,
    mode="a",
    header=not output_file.exists(),
    index=False,
)

Appending creates additional risks:

• The same batch may be loaded twice.
• Headers may be repeated.
• Partially loaded batches may be difficult to remove.
• Source records may later be updated or deleted.
• Schemas may change.

Production pipelines often use database transactions, merge operations, staging tables, partition replacement, or unique batch identifiers instead of blindly appending to a file.

Logging the ETL Pipeline

Logs create a chronological record of pipeline activity.

The example uses Python’s built-in logging module:

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(),
    ],
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE, encoding="utf-8"),
        logging.StreamHandler(),
    ],
)

This records messages in both:

• logs/etl.log
• The terminal

Python’s logging system provides timestamps, severity levels, exception reporting, formatters, and multiple output handlers. It is preferable to manually opening a text file for each message. Python Logging HOWTO

Example output:

2026-08-23 10:00:00 | INFO | ETL pipeline started
2026-08-23 10:00:00 | INFO | Reading CSV file: data/input/customers_01.csv
2026-08-23 10:00:00 | INFO | Extracted 3 rows from customers_01.csv
2026-08-23 10:00:01 | INFO | Extraction complete: 3 files and 8 rows
2026-08-23 10:00:01 | INFO | Transformation complete: 8 rows
2026-08-23 10:00:01 | INFO | Loaded 8 rows into data/output/people_metric.csv
2026-08-23 10:00:01 | INFO | ETL pipeline completed successfully

2026-08-23 10:00:00 | INFO | ETL pipeline started
2026-08-23 10:00:00 | INFO | Reading CSV file: data/input/customers_01.csv
2026-08-23 10:00:00 | INFO | Extracted 3 rows from customers_01.csv
2026-08-23 10:00:01 | INFO | Extraction complete: 3 files and 8 rows
2026-08-23 10:00:01 | INFO | Transformation complete: 8 rows
2026-08-23 10:00:01 | INFO | Loaded 8 rows into data/output/people_metric.csv
2026-08-23 10:00:01 | INFO | ETL pipeline completed successfully

If an exception occurs, logging.exception() records both the message and traceback:

except Exception:
    logging.exception("ETL pipeline failed")
    raise

except Exception:
    logging.exception("ETL pipeline failed")
    raise

Raising the exception again ensures that schedulers and calling systems can detect the failure.

Why Separate ETL Functions?

The program defines separate functions for:

• CSV extraction
• JSON extraction
• Schema validation
• Complete extraction
• Transformation
• Loading
• Pipeline execution

This separation improves:

• Readability
• Testing
• Reuse
• Troubleshooting
• Maintenance

For example, the conversion rules can be tested without reading any files:

def test_transform_converts_units():
    source = pd.DataFrame(
        {
            "name": ["Alice"],
            "height_inches": [65],
            "weight_pounds": [130],
            "source_file": ["test.csv"],
        }
    )

    result = transform(source)

    assert result.loc[0, "height_meters"] == 1.65
    assert result.loc[0, "weight_kilograms"] == 58.97

def test_transform_converts_units():
    source = pd.DataFrame(
        {
            "name": ["Alice"],
            "height_inches": [65],
            "weight_pounds": [130],
            "source_file": ["test.csv"],
        }
    )

    result = transform(source)

    assert result.loc[0, "height_meters"] == 1.65
    assert result.loc[0, "weight_kilograms"] == 58.97

Running the Program

Install pandas:

python -m pip install pandas

python -m pip install pandas

Run the pipeline from the project directory:

python etl.py

python etl.py

After successful execution, the output will be available at:

data/output/people_metric.csv

data/output/people_metric.csv

The execution log will be available at:

logs/etl.log

logs/etl.log

Expected Output

The target file might look like this:

name,height_meters,weight_kilograms,source_file
Alice,1.65,58.97,customers_01.csv
Bob,1.78,81.65,customers_01.csv
Carla,1.57,52.16,customers_01.csv
Daniel,1.83,90.72,customers_03.json
Elena,1.63,56.7,customers_03.json

name,height_meters,weight_kilograms,source_file
Alice,1.65,58.97,customers_01.csv
Bob,1.78,81.65,customers_01.csv
Carla,1.57,52.16,customers_01.csv
Daniel,1.83,90.72,customers_03.json
Elena,1.63,56.7,customers_03.json

Limitations of This Example

This program is appropriate for learning and relatively small collections of local files. It is not a complete production architecture.

Memory limitations

pandas loads each file and the combined result into memory. It may not be appropriate for datasets larger than the available memory.

Alternatives can include:

• Reading files in chunks
• Loading directly into a database
• Processing partitioned files
• Using a distributed processing engine
• Using warehouse-native transformations

No incremental-state management

The program processes every source file each time. A production pipeline may track:

• Previously processed files
• Source modification times
• Watermarks
• Transaction identifiers
• Batch identifiers
• Change-data-capture positions

Limited schema management

The example verifies required columns but does not enforce a complete versioned schema or reject unexpected fields.

No quarantine output

Invalid rows stop the complete pipeline. Many production systems store rejected records separately with error reasons.

CSV destination

CSV does not provide transactions, indexes, constraints, efficient type preservation, or concurrent writes. A database, warehouse, or columnar file format may be more appropriate for production use.

Simplified domain example

Height and weight conversion demonstrates ETL mechanics. These measurements alone should not be treated as sufficient inputs for a clinically valid diabetes-risk system.

ETL Versus ELT

Modern analytical platforms frequently use ELT:

Extract → Load → Transform

Extract → Load → Transform

In ELT, source data is loaded into a warehouse or lake before business transformations are applied.

Neither sequence is universally better. The correct approach depends on security, latency, cost, processing capabilities, governance, and analytical requirements.

Key Takeaways

• ETL stands for Extract, Transform, and Load.
• ETL commonly operates in batches but is not synonymous with batch processing.
• Extraction collects data and should preserve source context.
• Transformation standardizes, validates, combines, and enriches data.
• Loading writes the result to a target file or system.
• Source units and schemas must be explicitly defined.
• pd.concat() should be used instead of the removed DataFrame.append() method.
• Collecting DataFrames and concatenating them once is more efficient than iterative concatenation.
• ignore_index=True creates a new sequential index after combining records.
• JSON arrays and line-delimited JSON require different reading configurations.
• Logging should capture stage progress, record counts, failures, and tracebacks.
• Production pipelines require stronger controls for idempotency, partial failure, schema changes, data quality, and recovery.

Conclusion

A simple ETL pipeline demonstrates the foundation of many data-engineering systems. It collects data from multiple sources, validates their schemas, converts values into a standard representation, and loads a consistent result into a destination.

The Python example also illustrates several production-oriented principles: deterministic file discovery, source provenance, explicit validation, modular functions, structured logging, and safer target-file replacement.

As data volumes and operational requirements grow, the same ETL concepts can be implemented with databases, orchestration platforms, cloud services, distributed processing engines, and governed data repositories.

One-sentence summary: ETL converts data from multiple sources into a validated and standardized destination by executing extraction, transformation, and loading in a controlled, observable sequence.