Reference library

Data pipelines & processing

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

4 matches

Deduplicate events by ID within a window in Python

Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.

deduplication events heapq

Python

import heapq
from collections import defaultdict

def deduplicate_events(events, window_size):
    """Return events deduplicated by id, keeping newest within each sliding window."""
    # Index events by (timestamp, id) for deterministic ordering
    events_by_id = defaultdict(list)
    for ts, eid, *payload in events…

14 0 Open

Data pipelines & processing medium

How to Count Events by Minute with a Tumbling Window in Python

Group timestamps into fixed 60-second tumbling windows and count events per bucket using a dict.

datetime grouping time-window

Python

from collections import defaultdict
from datetime import datetime, timedelta


def tumbling_window_count(events, window_seconds=60):
    buckets = defaultdict(int)
    for event in events:
        ts = datetime.fromisoformat(event["timestamp"])
        bucket_start = ts - timedelta(seconds=ts.second % window_seconds,
…

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

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.