Aligning Tumbling Window Aggregations to Full Hours

[kriskonina]

I am using eKuiper to process streaming data from MQTT with the following payload format:

field	type
id	string
ts	datetime
val	bigint

Data is emitted every 15 seconds, and I am aggregating the sums of the val field over 1-hour tumbling windows based on the ts timestamp provided in each payload. I need the aggregation windows to align with full hours; for example, if streaming starts at 08:45, I want the first aggregation window to end at 09:00, not 09:45, effectively snapping the window start to align with full hours (08:00, 09:00, etc.).
Is there any known ways to achieve this?

[ngjaying]

@kriskonina Sounds like you want to use event time.

When creating stream, specify the timestamp field for it. Example like

{
  "sql": "create stream jsonDemo () WITH (TYPE=\"mqtt\", FORMAT=\"json\", SHARED=\"TRUE\", DATASOURCE=\"demo\", TIMESTAMP=\"ts\")"
}

In your rule, need to set its option isEventTime to true

{
  "id": "rule3",
  "sql": "SELECT * FROM jsonDemo GROUP BY TumblingWindow(hh, 1)",
  "actions": [
    {
      "mqtt": {
        "server": "tcp://127.0.0.1:1883",
        "topic": "result/rule3"
      }
    }
  ],
  "options": {
    "isEventTime": true,
    "lateTolerance": 0
  }
}

[kriskonina]

Appreciate your help, @ngjaying! It seems that setting lateTolerance to 0 (was previously set to 1000) did the trick in the end... Are these options documented anywhere? What does lateTolerance mean in this context?

[ngjaying]

Check here https://ekuiper.org/docs/en/latest/guide/rules/overview.html#fine-tuning