How to Aggregate Periodic Snapshot Data in Python
Generates mock snapshot data and groups values into periods to compute average aggregates with Python's standard library.
Python code
29 linesimport random
from collections import defaultdict
def snapshot_aggregate(n=10, period=3):
data = defaultdict(list)
for i in range(n):
key = f"item_{i % period}"
data[key].append(random.randint(1, 100))
return dict(data)
def aggregate_periodic(snapshots, period=3):
result = {}
for key, values in snapshots.items():
group_size = period
groups = []
for i in range(0, len(values), group_size):
chunk = values[i:i + group_size]
if len(chunk) == group_size:
groups.append(sum(chunk) / len(chunk))
if groups:
result[key] = groups
return result
if __name__ == "__main__":
random.seed(42)
snapshots = snapshot_aggregate()
periodic_agg = aggregate_periodic(snapshots)
print("Snapshots:", dict(snapshots))
print("Periodic averages (period=3):", periodic_agg)Output
Snapshots: {'item_0': [24, 79, 14, 86, 4], 'item_1': [21, 88, 44, 58], 'item_2': [81, 7, 36]}
Periodic averages (period=3): {'item_0': [39.0], 'item_1': [51.0]}How it works
The snapshot_aggregate function emulates incoming periodic data by assigning random values to keys based on a cycle, mimicking a message stream. The aggregate_periodic function slices each key's values into fixed-size chunks using Python's slicing syntax, then computes the average only for complete periods — ignoring incomplete trailing values. Using defaultdict(list) simplifies appending without checking key existence. The if __name__ == "__main__" guard keeps the demonstration isolated when the module is imported elsewhere.
Common mistakes
- Including incomplete chunks in the aggregation instead of discarding them
- Forgetting to seed the random number generator for reproducible output
- Confusing the period parameter with the number of keys generated
- Mutating the original snapshot dict while iterating it inside aggregate_periodic
Variations
- Use `statistics.fmean(chunk)` for faster float-mean computations on large datasets
- Aggregate with `collections.deque` for sliding-window stream processing without storing all values
Real-world use cases
- Batch processing time-series metric snapshots from Kafka topics into 5-minute averages for dashboards.
- Computing rolling user-engagement scores by grouping session events into fixed-hour windows.
- Condensing raw IoT sensor readings into per-minute averages before pushing to a data warehouse.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.