Reference library

Data pipelines & processing

ETL-style flows, batch transforms, validation, and moving data between formats.

5 matches

ETL in Python: Extract CSV, Transform Dict, Load JSON

Build a simple ETL pipeline in Python that reads a CSV file, transforms each row (stripping whitespace and converting numeric fields), and writes the result to JSON.

etl csv json

Python

import csv
import json
from pathlib import Path

def extract_csv(file_path):
    """Read CSV file and return list of row dictionaries."""
    with Path(file_path).open('r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        return list(reader)

def transform_dicts(rows):
    """Transform ro…

14 0 Open

Data pipelines & processing easy

ETL in Python: Extract CSV, Transform Dicts, Load JSON

Build a simple ETL pipeline that reads a CSV, normalizes keys and converts price to float, then writes structured JSON.

etl csv json

Python

import csv
import json
from pathlib import Path

def etl_csv_to_json(csv_path: str, json_path: str) -> None:
    """Extract CSV, transform rows to dicts, load to JSON."""
    with open(csv_path, mode='r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        records = list(reader)

    # Trans…

12 0 Open

Data pipelines & processing easy

How to Parse Data in Python: A Beginner's Helper

This helper parses a JSON payload, extracts user names, emails, and signup dates, then summarizes the results.

json parsing data-processing

Python

import json
from datetime import datetime
from typing import Dict, List


def parse_data(payload: str) -> Dict[str, List]:
    """Parse a JSON payload and extract useful fields."""
    raw = json.loads(payload)
    users = raw.get("users", [])

    parsed = {
        "names": [],
        "emails": [],
        "signup_…

16 0 Open

Data pipelines & processing easy

How to Partition Output Files by Date Key in Python

Group output files into a dictionary partitioned by a YYYYMMDD date key extracted from the filename prefix.

file-partitioning date-key pathlib

Python

from pathlib import Path
from collections import defaultdict

def partition_files_by_date(directory: str) -> dict:
    """Partition output files by date key extracted from filename (YYYYMMDD prefix)."""
    path = Path(directory)
    partitions = defaultdict(list)
    
    for file in path.iterdir():
        if file.i…

14 0 Open

Data pipelines & processing easy

Parallel Extract Multiple Sources with Threads in Python

Extract data from multiple sources in parallel using ThreadPoolExecutor and verify results match sequential processing.

threads threadpoolexecutor concurrency

Python

import threading
from concurrent.futures import ThreadPoolExecutor

def extract_from_source(source):
    """Simulate extracting data from a source."""
    return f"Data from {source}"

def main():
    sources = ["source_a", "source_b", "source_c", "source_d"]
    
    # Sequential extraction for comparison
    sequent…

14 0 Open

Browse by section

Each section groups closely related Python snippets.