Reference library

Data pipelines & processing

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

3 matches

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 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

Browse by section

Each section groups closely related Python snippets.