Reference library

Data pipelines & processing

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

6 matches

Enrich a stream with reference data by key lookup in Python

Uses streamz to join each incoming record to a reference dictionary by name, adding department and level fields or defaults.

streamz streaming join

Python

from streamz import Stream

reference = {"alice": {"dept": "eng", "level": 3}, "bob": {"dept": "sales", "level": 5}}

def enrich(record):
    name = record.get("name")
    ref = reference.get(name)
    joined = dict(record)
    if ref:
        joined.update(ref)
    else:
        joined["dept"] = "unknown"
        joi…

13 0 Open

Data pipelines & processing easy

How to Implement a Sliding Window Average in Python

Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.

deque sliding-window streaming

Python

from collections import deque


class SlidingWindowAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque(maxlen=window_size)
        self.total = 0

    def add(self, value):
        if len(self.window) == self.window_size:
            self.total -= self.windo…

15 0 Open

Data pipelines & processing medium

How to Stream a Large JSONL File Line by Line in Python

Process a large JSON-lines file incrementally using streaming techniques to avoid loading the entire file into memory.

streaming jsonl large-files

Python

import json

def process_large_file(filepath, chunk_size=8192):
    """
    Stream a large JSON-lines file line by line, processing each record
    without loading the entire file into memory.
    """
    total_count = 0
    total_sum = 0
    
    with open(filepath, 'r') as f:
        while True:
            chunk = …

13 0 Open

Data pipelines & processing easy

How to Track Checkpoint Offset After Batch Commit in Python

A batch processor that tracks the last successfully committed offset after processing records in batches, advancing the checkpoint only when each batch commits successfully.

batch-processing checkpoint offset

Python

import json
from typing import Any


class BatchProcessor:
    """Tracks checkpoint offset after committing batches."""

    def __init__(self, batch_size: int = 3):
        self.batch_size = batch_size
        self.offset = 0  # last successfully committed offset (exclusive)
        self.total_committed = 0

    def …

12 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

Implement an Out-of-Order Sort Buffer with a Heap in Python

Buffers out-of-order indices from a stream and emits them in sorted order using a min-heap with a sliding window.

heapq sorting streaming

Python

import heapq
from collections import deque


class OutOfOrderSorter:
    def __init__(self, buffer_size):
        self.buffer_size = buffer_size
        self.buffer = deque(maxlen=buffer_size)
        self.heap = []
        self.next_expected_index = 0
        self.result = []

    def push(self, item):
        heapq.…

12 0 Open

Browse by section

Each section groups closely related Python snippets.