Apache Flink：ProcessWindowFunction 不适用答案

【问题标题】：Apache Flink: ProcessWindowFunction is not applicableApache Flink：ProcessWindowFunction 不适用
【发布时间】：2018-03-20 05:44:36
【问题描述】：

我想在我的 Apache Flink 项目中使用 ProcessWindowFunction。但是我在使用进程函数时遇到了一些错误，请参见下面的代码 sn-p

错误是：

WindowedStream,Tuple,TimeWindow> 类型中的方法 process(ProcessWindowFunction,R,Tuple,TimeWindow>) 不适用于参数 (JDBCExample.MyProcessWindows)

我的程序：

DataStream<Tuple2<String, JSONObject>> inputStream;

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

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

我的ProcessWindowFunction：

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 
  {
    ...
  }

}

【问题讨论】：

你能仔细检查错误信息吗？似乎process() 方法的签名中缺少某些内容。

标签： apache-flink flink-streaming stream-processing

【解决方案1】：

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

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

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

通过将String 替换为Tuple，您现在有了一个通用的键占位符，当您需要在processElement() 方法中访问键时，您可以将其转换为Tuple1<String>：

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<IN, KEY> 函数来提取密钥，则可以避免强制转换并使用正确的类型，因为编译器知道KeySelector 的返回类型KEY。

【讨论】：

感谢@Fabian 的帮助，我将 String 替换为 Tuple 但仍然存在相同的错误。我是否也需要定义 KeySelector 函数。是强制性的吗？？像 apply 这样的简单窗口操作也会抛出相同的错误，而 reduce 工作正常。

【解决方案2】：

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 like：

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

【讨论】：