Apache Flink：ProcessWindowFunction不适用

Question

我想在Apache Flink项目中使用ProcessWindowFunction。但我在使用process函数时遇到了一些错误，请参见下面的代码片段

错误是：

方法进程（ProcessWindowFunction， R， Tuple， TimeWindow

我的程序：

DataStream<Tuple2<String, JSONObject>> inputStream;

inputStream = env.addSource(new JsonArraySource());

inputStream.keyBy(0)
  .window(TumblingEventTimeWindows.of(Time.minutes(10)))
  .process(new MyProcessWindows());

我的处理窗口功能：

private class MyProcessWindows 
  extends ProcessWindowFunction<Tuple2<String, JSONObject>, Tuple2<String, String>, String, Window>
{

  public void process(
      String key, 
      Context context, 
      Iterable<Tuple2<String, JSONObject>> input, 
      Collector<Tuple2<String, String>> out) throws Exception 
  {
    ...
  }

}

Answer 1

Fabian说：）使用Tuple应该可以工作，但确实会在您的ProcessWindowFunction中涉及一些丑陋的类型转换。使用KeySelector很容易，并且会产生更简洁的代码。例如。

.keyBy(new KeySelector<Tuple2<String,JsonObject>, String>() {

    @Override
    public String getKey(Tuple2<String, JsonObject> in) throws Exception {
        return in.f0;
    }
})

然后，通过上面的操作，您可以定义一个ProcessWindowFunction，如：

public class MyProcessWindows extends ProcessWindowFunction<Tuple2<String, JsonObject>, Tuple2<String, String>, String, TimeWindow> {

Answer 2

问题可能是ProcessWindowFunction的泛型类型。

您正在按位置引用按键（keyBy（0））。因此，编译器无法推断其类型（字符串），您需要将ProcessWindowFunction更改为：

private class MyProcessWindows 
    extends ProcessWindowFunction<Tuple2<String, JSONObject>, Tuple2<String, String>, Tuple, Window>

通过将String替换为Tuple，您现在有了一个可以强制转换为Tuple1的键的通用占位符

public void process(
    Tuple key, 
    Context context, 
    Iterable<Tuple2<String, JSONObject>> input, 
    Collector<Tuple2<String, String>> out) throws Exception {

  String sKey = (String)((Tuple1)key).f0;
  ...
}

如果您定义了KeySelector，您可以避免强制转换并使用正确的类型