在flink中使用json的注意事项：
使用 JSONDeserializationSchema 对事件进行反序列化 ObjectNode s。你可以绘制Map ObjectNode 至 YourObject 为了方便或继续与 ObjectNode .
使用的教程 ObjectNode : http://www.baeldung.com/jackson-json-node-tree-model
回到您的案例，您可以按以下方式进行：

val eventStream : DataStream [ObjectNode] = 
oneMinuteAgg
    .addSource(source)
    .windowAll()
    .TimeWindow(Time.minutes(1))
    .trigger(new MyTriggerFunc)
    .aggregation(new MyAggregationFunc)

将输出1分钟的聚合流

[     
      {
      "date" :2018-03-03
      "sum" : 120
      }, 
      {
      "date" :2018-03-03
      "sum" : 120
      }
]

然后将另一个操作符链接到“oneminuteagg”，该操作符将把1分钟的聚合添加到1天的聚合中：

[...]
oneMinuteAgg
        .windowAll()
        .TimeWindow(Time.days(1))
        .trigger(new Whatever)
        .aggregation(new YourDayAggF)

这将输出您所需要的

{
    "aggregationType" : "day"
    "days before" : 4
    "aggregates : [{
      "date" :2018-03-03
      "sum" : 120
      }, 
      {
      "date" :2018-03-03
      "sum" : 120
      }]
}

我曾经 windowAll() 假设您不需要为流设置密钥。