Skip to main content
Temporal stage showing time-window grouping with drift detection
The Temporal stage groups documents into time windows (hour, day, week, month, quarter, year) and computes aggregations per window. It can also detect drift between consecutive windows, flagging spikes or drops that exceed a threshold.
Stage Category: REDUCE (Groups results by time)Transformation: N documents → M time-window results (with optional drift detection)

​
When to Use

Use CaseDescription
Trend analysisTrack how metrics change over time
Spike detectionFlag windows where a metric jumps or drops significantly
Content velocityCount new documents per day/week/month
Temporal distributionUnderstand when content was created or modified

​
When NOT to Use

ScenarioRecommended Alternative
Simple countingaggregate
Full time-series from databaseAggregation API
Grouping by non-time fieldsgroup_by
LLM-based trend analysissummarize

​
Parameters

ParameterTypeDefaultDescription
time_fieldstringRequiredDocument field containing the timestamp
windowstringRequiredWindow granularity: hour, day, week, month, quarter, year
aggregationsarrayRequiredList of aggregation operations per window
driftobjectnullDrift detection configuration
sort_orderstring"asc"Sort windows asc (oldest first) or desc (newest first)
limitintegernullMax number of windows to return
include_documentsbooleanfalseInclude original documents in output

​
Aggregation Types

TypeField RequiredDescription
countNoNumber of documents in window
sumYesSum of field values
avgYesAverage value
minYesMinimum value
maxYesMaximum value
count_distinctYesUnique values count
collect_distinctYesGather unique values into list

​
Drift Detection

Drift detection compares a metric between consecutive windows and computes the percent change.
ParameterTypeDefaultDescription
drift.enabledbooleanfalseEnable drift detection
drift.metricstringRequired if enabledWhich aggregation alias to track
drift.thresholdfloatnullPercent change to flag (e.g., 50.0 flags changes > 50%)

​
Timestamp Formats

The stage parses timestamps automatically:
FormatExample
ISO 8601"2026-04-01T10:30:00Z"
ISO date"2026-04-01"
Epoch seconds (int)1743465600
Epoch seconds (float)1743465600.123
Documents with missing or unparseable timestamps are skipped (counted in num_documents_skipped).

​
Configuration Examples

{
  "stage_type": "reduce",
  "stage_id": "temporal",
  "parameters": {
    "time_field": "created_at",
    "window": "day",
    "aggregations": [
      {"function": "count", "alias": "posts_per_day"},
      {"function": "avg", "field": "score", "alias": "avg_relevance"}
    ]
  }
}

​
Output Schema

​
Window Results

{
  "metadata": {
    "windows": [
      {
        "window": "2026-04-01",
        "metrics": {
          "count": 3,
          "avg_score": 0.85
        }
      },
      {
        "window": "2026-04-02",
        "metrics": {
          "count": 1,
          "avg_score": 0.72
        },
        "drift": {
          "absolute_change": -2,
          "percent_change": -66.67,
          "flagged": true
        }
      }
    ],
    "num_windows": 2,
    "num_documents_in": 4,
    "num_documents_skipped": 0,
    "window_granularity": "day"
  }
}

​
Window Key Formats

WindowFormatExample
hourYYYY-MM-DDTHH:00:002026-04-01T15:00:00
dayYYYY-MM-DD2026-04-01
weekYYYY-WNN2026-W14
monthYYYY-MM2026-04
quarterYYYY-QN2026-Q2
yearYYYY2026

​
Performance

MetricValue
Latency5-50ms
MemoryO(windows x aggregations)
CostFree
ScalabilityEfficient for large result sets

​
Common Pipeline Patterns

​
Search + Temporal Analysis

[
  {
    "stage_type": "filter",
    "stage_id": "feature_search",
    "parameters": {
      "searches": [
        {
          "feature_uri": "mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding",
          "query": "{{INPUT.query}}",
          "top_k": 500
        }
      ],
      "final_top_k": 500
    }
  },
  {
    "stage_type": "reduce",
    "stage_id": "temporal",
    "parameters": {
      "time_field": "created_at",
      "window": "day",
      "aggregations": [
        {"function": "count", "alias": "matches_per_day"},
        {"function": "avg", "field": "score", "alias": "avg_relevance"}
      ],
      "drift": {
        "enabled": true,
        "metric": "matches_per_day",
        "threshold": 100.0
      }
    }
  }
]

​
Brand Monitoring with Spike Detection

[
  {
    "stage_type": "filter",
    "stage_id": "feature_search",
    "parameters": {
      "searches": [
        {
          "feature_uri": "mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding",
          "query": "{{INPUT.brand_name}}",
          "top_k": 1000
        }
      ],
      "final_top_k": 1000
    }
  },
  {
    "stage_type": "enrich",
    "stage_id": "llm_enrich",
    "parameters": {
      "provider": "google",
      "model_name": "gemini-2.5-flash-lite",
      "prompt": "Classify sentiment as positive, neutral, or negative: {{DOC.content}}",
      "output_field": "sentiment"
    }
  },
  {
    "stage_type": "reduce",
    "stage_id": "temporal",
    "parameters": {
      "time_field": "published_at",
      "window": "week",
      "aggregations": [
        {"function": "count", "alias": "mentions"},
        {"function": "count_distinct", "field": "sentiment", "alias": "sentiment_spread"}
      ],
      "drift": {
        "enabled": true,
        "metric": "mentions",
        "threshold": 50.0
      }
    }
  }
]

​
Error Handling

ErrorBehavior
Missing timestamp fieldDocument skipped
Unparseable timestampDocument skipped
Non-numeric field for sum/avgDocument skipped for that aggregation
Empty results0 windows returned
Unknown aggregationReturns null
  • Aggregate - Statistical aggregations without time grouping
  • Group By - Group documents by any field
  • Sample - Statistical sampling