Data Aggregator¶

Collect and process data from workflow steps over configurable time-based or run-based intervals to generate statistical summaries and analytics reports, supporting multiple aggregation operations (sum, average, max, min, count, distinct values, value counts) with optional UQL-based data transformations for comprehensive data stream analytics.

How This Block Works¶

This block collects and aggregates data from workflow steps over specified intervals to produce statistical summaries. Unlike most blocks that output data for every input, this block maintains internal state and outputs aggregated results only when the configured interval is reached. The block:

1. Receives data inputs from other workflow steps (via data field mapping variable names to workflow step outputs)
2. Optionally applies UQL (Query Language) operations to transform the data before aggregation (e.g., extract class names from detections, calculate sequence lengths, filter or transform values) using data_operations for each input variable
3. Accumulates data into internal aggregation states based on the specified aggregation_mode for each variable
4. Tracks time elapsed or number of runs based on interval_unit (seconds, minutes, hours, or runs)
5. Most of the time, returns empty outputs (terminating downstream processing) while collecting data internally
6. When the interval threshold is reached (based on time elapsed or run count), computes and outputs aggregated statistics
7. Flushes internal state after outputting aggregated results and starts collecting data for the next interval
8. Produces output fields dynamically named as {variable_name}_{aggregation_mode} (e.g., predictions_avg, classes_distinct, count_values_counts)

The block supports multiple aggregation modes for numeric data (sum, avg, max, min, values_difference), counting operations (count, count_distinct), and value analysis (distinct, values_counts). For list-like data, operations automatically process each element (e.g., count adds list length, distinct adds each element to the distinct set). The interval can be time-based (useful for video streams where wall-clock time matters) or run-based (useful for video file processing where frame count matters more than elapsed time).

Common Use Cases¶

• Video Stream Analytics: Aggregate detection results over time intervals from live video streams (e.g., calculate average object counts per minute, track distinct classes seen per hour, compute min/max detection counts over 30-second windows), enabling real-time analytics and monitoring for continuous video processing workflows
• Batch Video Processing: Aggregate statistics across video frames using run-based intervals (e.g., calculate average detections per 100 frames, count distinct objects across 500-frame windows, sum total detections per batch), enabling meaningful analytics for pre-recorded video files where frame count matters more than elapsed time
• Time-Series Metrics Collection: Collect and summarize workflow metrics over time (e.g., aggregate detection counts, calculate average confidence scores, track distinct class occurrences, compute value distributions), enabling statistical analysis and reporting for production workflows
• Model Performance Analysis: Analyze model predictions across multiple inputs (e.g., calculate average prediction counts, track distinct predicted classes, compute min/max confidence scores, count occurrences of each class), enabling comprehensive model performance evaluation and insights
• Data Stream Summarization: Summarize high-frequency data streams into periodic reports (e.g., aggregate every 60 seconds of detections into summary statistics, compute hourly averages, generate per-run summaries), enabling efficient data reduction and analysis for high-volume workflows
• Multi-Model Comparison: Aggregate results from multiple models for comparison (e.g., compare average detection counts across models, track distinct classes per model, compute aggregate statistics for model ensembles), enabling comparative analytics across different inference pipelines

Connecting to Other Blocks¶

This block receives data from workflow steps and outputs aggregated statistics periodically:

• After detection or analysis blocks (e.g., Object Detection, Instance Segmentation, Classification) to aggregate prediction results over time or across frames, enabling statistical analysis of model outputs and detection patterns
• After data processing blocks (e.g., Expression, Property Definition, Detections Filter) that produce numeric or list outputs to aggregate computed values, metrics, or transformed data over intervals
• Before sink blocks (e.g., CSV Formatter, Local File Sink, Webhook Sink) to save periodic aggregated reports, enabling efficient storage and export of summarized analytics data instead of individual data points
• In video processing workflows to generate time-based or frame-based analytics reports, enabling comprehensive video analysis with periodic statistical summaries rather than per-frame outputs
• Before visualization or reporting blocks that need aggregated data to create dashboards, charts, or summaries from time-series data, enabling visualization of trends and statistics
• In analytics pipelines where high-frequency data needs to be reduced to periodic summaries, enabling efficient downstream processing and storage of statistical insights rather than raw high-volume data streams

Type identifier¶

Use the following identifier in step "type" field: roboflow_core/data_aggregator@v1to add the block as as step in your workflow.

Properties¶

The Refs column marks possibility to parametrise the property with dynamic values available in workflow runtime. See Bindings for more info.

Available Connections¶

Input and Output Bindings¶

The available connections depend on its binding kinds. Check what binding kinds Data Aggregator in version v1 has.

{
    "name": "<your_step_name_here>",
    "type": "roboflow_core/data_aggregator@v1",
    "data": {
        "predictions": "$steps.model.predictions",
        "reference": "$inputs.reference_class_names"
    },
    "data_operations": {
        "predictions": [
            {
                "property_name": "class_name",
                "type": "DetectionsPropertyExtract"
            }
        ]
    },
    "aggregation_mode": {
        "predictions": [
            "distinct",
            "count_distinct"
        ]
    },
    "interval_unit": "seconds",
    "interval": 10
}