如何在Cumulocity CEP查询中分组历史数据？按天分组近一周数据

看起来你在使用Esper处理时序数据，想要按天分组最近一周的Avg_Load数据，但现有代码的问题在于没有正确拆分批量拉取的集合，而且用了当前时间的日期而非测量数据本身的时间来分组，导致没法得到正确的每日分组结果。我来帮你修正一下：

首先拆解核心问题：你每10秒拉取的是7天内的历史数据集合，直接插入表的话无法对单个测量数据做分组；另外你插入时用的current_timestamp().toDate()是当前执行时间的日期，不是每条测量数据的时间，这显然不是你需要的分组依据。

修正方案1：直接按天聚合查询 #

先把批量拉取的集合拆成单个事件，再按测量数据的日期分组聚合：

// 先定义测量事件的Schema（如果没提前定义的话）
create schema MeasurementEvent (load double, eventTime timestamp);

// 每10秒拉取数据，用unwind把集合拆成单个事件流入MeasurementStream
insert into MeasurementStream
select m.load as load, m.eventTime as eventTime
from pattern[every timer:interval(10 sec)]
unwind findAllMeasurementByFragmentTypeAndSourceAndTimeBetween('Avg_Load','764697', current_timestamp().minus(7 day), current_timestamp()) as m;

// 按天分组，计算每日的聚合结果（这里用平均负载和数据量为例，你可以换成sum/max等）
@Name("Occupancy") @Resilient
select
    date_format(eventTime, 'yyyy-MM-dd') as day,
    avg(load) as dailyAvgLoad,
    count(*) as totalMeasurements
from MeasurementStream
group by date_format(eventTime, 'yyyy-MM-dd');

修正方案2：将分组结果持久化到表 #

如果需要把每日分组结果存入表方便后续查询，可以添加插入逻辑：

insert into DailyMeasurements
select
    date_format(eventTime, 'yyyy-MM-dd') as day,
    avg(load) as dailyAvgLoad,
    count(*) as totalMeasurements
from MeasurementStream
group by date_format(eventTime, 'yyyy-MM-dd');

// 之后直接从DailyMeasurements查询每日数据
@Name("Occupancy") @Resilient
select * from DailyMeasurements;

优化建议：增量拉取避免重复数据 #

你现在每10秒拉取全7天的数据，会导致大量重复数据流入流中，效率很低。可以用变量记录上次拉取的时间，每次只拉取新增的数据：

// 初始化上次拉取时间为7天前
create variable timestamp lastPullTime = current_timestamp().minus(7 day);

// 每次拉取从lastPullTime到当前时间的数据，拉取后更新lastPullTime
insert into MeasurementStream
select m.load as load, m.eventTime as eventTime
from pattern[every timer:interval(10 sec)]
unwind findAllMeasurementByFragmentTypeAndSourceAndTimeBetween('Avg_Load','764697', lastPullTime, current_timestamp()) as m
set lastPullTime = current_timestamp();

关键注意点 #

• unwind是核心：它能把findAll返回的集合拆分成单个事件，这样才能对每条测量数据做时间分组。
• 分组依据必须是测量数据本身的时间（比如示例中的m.eventTime，你需要根据实际返回的对象字段调整），而非当前执行插入的时间。
• 聚合函数（avg/count等）可根据你的实际需求替换，比如要统计每日总负载就用sum(load)。

内容的提问来源于stack exchange，提问作者Joby