Reference library

Data pipelines & processing

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

4 matches

How to Compress Pipeline Output Gzip Per Partition in Python

Compress each partition of pipeline output into a separate gzip file and verify the compressed data by reading it back.

import gzip
import io
import random
from pathlib import Path


def compress_partition(partition_data: list[str], output_path: Path) -> int:
    """Compress a partition of data to a gzip file, returns bytes written."""
    with gzip.open(output_path, 'wt', encoding='utf-8') as f:
        f.writelines(partition_data)
  …

13 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

How to route late-arriving data to a side output in Python

Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.

data pipelines streaming dead-letter

Python

from collections import defaultdict

def late_arriving_side_output(events, late_threshold_ts):
    """
    Mock a streaming pipeline that separates late-arriving data events
    into a side output list (e.g., for dead-letter analysis).

    events: list of (timestamp, data) tuples, timestamps as ints.
    late_thresho…

12 0 Open

Data pipelines & processing medium

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel

Python

from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …

12 0 Open

Browse by section

Each section groups closely related Python snippets.