Flink 窗口:聚合并输出到接收器

Question

我们有一个数据流，其中每个元素都是这样的类型：

id: String
type: Type
amount: Integer

我们希望聚合此流并每周输出一次amount的总和。

当前解决方案：

一个示例 flink 管道如下所示：

stream.keyBy(type)
.window(TumblingProcessingTimeWindows.of(Time.days(7)))
.reduce(sumAmount())
.addSink(someOutput())

对于输入

| id | type | amount |
| 1  | CAT  | 10     |
| 2  | DOG  | 20     |
| 3  | CAT  | 5      |
| 4  | DOG  | 15     |
| 5  | DOG  | 50     |

如果窗口在记录3和4之间结束，我们的输出将是：

| TYPE | sumAmount |
| CAT  | 15        | (id 1 and id 3 added together)
| DOG  | 20        | (only id 2 as been 'summed')

Id4和5仍将在 flink 管道内，并将在下周输出。

因此，下周我们的总产出将是：

| TYPE | sumAmount |
| CAT  | 15        | (of last week)
| DOG  | 20        | (of last week)
| DOG  | 65        | (id 4 and id 5 added together)

新要求：

我们现在还想知道每条记录在哪一周处理了每条记录。换句话说，我们的新输出应该是：

| TYPE | sumAmount | weekNumber |
| CAT  | 15        | 1          |
| DOG  | 20        | 1          |
| DOG  | 65        | 2          |

但我们也想要一个额外的输出，如下所示：

| id | weekNumber |
| 1  | 1          |
| 2  | 1          |
| 3  | 1          |
| 4  | 2          |
| 5  | 2          |

如何处理？

flink 有什么办法可以做到这一点吗？我想我们会有一个聚合函数，例如对金额求和，但也输出每条记录与当前周数，但我在文档中找不到做到这一点的方法。

(注意：我们每周处理大约 1 亿条记录，所以理想情况下，我们只想在一周内将聚合保持在 Flink 的状态，而不是所有单独的记录(

编辑：

我选择了安东在下面描述的解决方案：

DataStream<Element> elements = 
stream.keyBy(type)
.process(myKeyedProcessFunction());
elements.addSink(outputElements());
elements.getSideOutput(outputTag)
.addSink(outputAggregates())

keyedProcessFunction看起来像这样：

class MyKeyedProcessFunction extends KeyedProcessFunction<Type, Element, Element>
private ValueState<ZonedDateTime> state;
private ValueState<Integer> sum;
public void processElement(Element e, Context c, Collector<Element> out) {
if (state.value() == null) {
state.update(ZonedDateTime.now());
sum.update(0);
c.timerService().registerProcessingTimeTimer(nowPlus7Days);
}
element.addAggregationId(state.value());
sum.update(sum.value() + element.getAmount());
}
public void onTimer(long timestamp, OnTimerContext c, Collector<Element> out) {
state.update(null);
c.output(outputTag, sum.value()); 
}
} 

Answer 1

reduce方法有一个变体，它将ProcessWindowFunction作为第二个参数。你会像这样使用它：

stream.keyBy(type)
.window(TumblingProcessingTimeWindows.of(Time.days(7)))
.reduce(sumAmount(), new WrapWithWeek())
.addSink(someOutput())
private static class WrapWithWeek
extends ProcessWindowFunction<Event, Tuple3<Type, Long, Long>, Type, TimeWindow> {
public void process(Type key,
Context context,
Iterable<Event> reducedEvents,
Collector<Tuple3<Type, Long, Long>> out) {
Long sum = reducedEvents.iterator().next();
out.collect(new Tuple3<Type, Long, Long>(key, context.window.getStart(), sum));
}
}

通常，ProcessWindowFunction 会传递一个包含窗口收集的所有事件的可迭代函数，但如果使用归约或聚合函数来预聚合窗口结果，则只有该单个值会传递到可迭代对象中。这方面的文档在这里，但文档中的示例目前有一个小错误，我已经在这里的示例中修复了该错误。

但考虑到对第二个输出的新要求，我建议你放弃用Windows执行此操作的想法，而是使用键控ProcessFunction。您需要两个每个键的 ValueState：一个按周计数，另一个用于存储总和。您需要一个每周触发一次的计时器：当它触发时，它应该发出类型、总和和周数，然后递增周数。同时，流程元素方法将简单地输出每个传入事件的 ID 以及周计数器的值。