Reference library

Data pipelines & processing

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

3 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

Browse by section

Each section groups closely related Python snippets.