Apache Flink-从另一个Stream的MapFunction访问WindowedStream的内部缓冲区

Question

我有一个基于Apache Flink的流媒体应用程序，具有以下设置：

< li >数据源:每分钟生成数据。 < li >使用CountWindow的窗口流，size=100，slide=1(滑动计数窗口)。 < li>ProcessWindowFunction对窗口中的数据应用一些计算(比如F(x))。 < li >使用输出流的数据接收器

这很好。现在，我想让用户能够提供一个函数G(x ),并将其应用于窗口中的当前数据，并将输出实时发送给用户

我不是在问如何应用任意函数G（x） - 我正在使用动态脚本来做到这一点。我正在询问如何从另一个流的映射函数访问窗口中的缓冲数据。

一些代码来澄清

DataStream<Foo> in  = .... // source data produced every minute
    in
       .keyBy(new MyKeySelector())
       .countWindow(100, 1)
       .process(new MyProcessFunction())
       .addSink(new MySinkFunction())

// The part above is working fine. Note that windowed stream created by countWindow() function above has to maintain internal buffer. Now the new requirement

DataStream<Function> userRequest  = .... // request function from user

userRequest.map(new MapFunction<Function, FunctionResult>(){
   public FunctionResult map(Function Gx) throws Exception {
         Iterable<Foo> windowedDataFromAbove = // HOW TO GET THIS???
         FunctionResult result = Gx.apply(windowedDataFromAbove);
         return result;

   }

})

Answer 1

假设Fx动态聚合传入的FOO，Gx处理一个窗口的FOO值，您应该能够实现以下目标：

DataStream<Function> userRequest  = .... // request function from user
Iterator<Function> iter = DataStreamUtils.collect(userRequest);
Function Gx = iter.next();

DataStream<Foo> in  = .... // source data
 .keyBy(new MyKeySelector())
 .countWindow(100, 1)
 .fold(new ArrayList<>(), new MyFoldFunc(), new MyProcessorFunc(Gx))
 .addSink(new MySinkFunction())

Fold函数（一旦传入的数据到达就对它们进行操作）可以这样定义：

private static class MyFoldFunc implements FoldFunction<foo, Tuple2<Integer, List<foo>>> {
    @Override
    public Tuple2<Integer, List<foo>> fold(Tuple2<Integer, List<foo>> acc, foo f) {
        acc.f0 = acc.f0 + 1; // if Fx is a simple aggregation (count)
        acc.f1.add(foo);
        return acc;
    }
}

处理器功能可以是这样的：

public class MyProcessorFunc
    extends ProcessWindowFunction<Tuple2<Integer, List<foo>>, Tuple2<Integer, FunctionResult>, String, TimeWindow> {

    public MyProcessorFunc(Function Gx) {
        super();
        this.Gx = Gx;
    }

    @Override
    public void process(String key, Context context,
                        Iterable<Tuple2<Integer, List<foo>> accIt,
                        Collector<Tuple2<Integer, FunctionResult>> out) {
        Tuple2<Integer, List<foo> acc = accIt.iterator().next();
        out.collect(new Tuple2<Integer, FunctionResult>(
            acc.f0, // your Fx aggregation
            Gx.apply(acc.f1), // your Gx results
        ));
    }
}

请注意，默认情况下，fold\reduce函数不在内部缓冲元素。我们在这里使用fold来计算动态指标，并创建一个窗口项列表。

如果您对在翻滚窗口(不是滑动的)上应用Gx感兴趣，您可以在管道中使用翻滚窗口。为了也计算滑动计数，可以有另一个只计算滑动计数的管道分支(不应用Gx)。这样，您不必在每个窗口中保存100个列表。

注意:您可能需要添加以下依赖项才能使用DataStreamUtils:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-contrib</artifactId>
    <version>0.10.2</version>
</dependency>

Answer 2

连接两个流，然后使用CoProcessFunction。获取Functions流的方法调用可以将它们应用于其他方法调用窗口中的内容。

如果您想要广播函数，那么您需要使用Flink 1.5（它支持连接键控流和广播流），或者使用一些直升机特技来创建一个可以包含Foo和函数类型的流，并适当复制函数（和密钥生成）以模拟广播。