无法使用 WindowStream.apply() 函数应用 WindowFunction [英] WindowFunction cannot be applied using WindowStream.apply() function

问题描述

我对使用 Apache Flink 和 Scala 还比较陌生，我刚刚开始掌握一些基本功能.我在尝试实现自定义 WindowFunction 时遇到了障碍.

I'm relatively new to using Apache Flink and Scala, and am just getting to grips with some of the basic functionality. I've hit a wall trying to implement a custom WindowFunction.

问题是，当我尝试实现自定义 WindowFunction 时，IDE 在.apply()"函数上给出错误

The problem is that when I try to implement a custom WindowFunction the IDE gives an error on the ".apply()" function

Cannot resolve symbol apply

Unspecified value parameters: foldFunction: (NotInferedR, Data.Fingerprint) => NotInferedR, windowFunction: (Tuple, TimeWindow, Iterable[NotInferedR], Collector[NotInferedR]) => Unit 
Unspecified value parameters: foldFunction: FoldFunction[Data.Fingerprint, NotInferedR], function: WindowFunction[NotInferedR, NotInferedR, Tuple, TimeWindow] 
Unspecified value parameters: function: WindowFunction[Data.Fingerprint, NotInferedR, Tuple, TimeWindow] 
Unspecified value parameters: windowFunction: (Tuple, TimeWindow, Iterable[Data.Fingerprint], Collector[NotInferedR]) => Unit 

Type mismatch, expected: (Tuple, TimeWindow, Iterable[Data.Fingerprint], Collector[NotInferedR]) => Unit, actual: DataTimeWindow.DataWindow 
Type mismatch, expected: WindowFunction[Data.Fingerprint, NotInferedR, Tuple, TimeWindow], actual: DataTimeWindow.DataWindow

这是我的代码:

val test = hashMap
      .keyBy("hash")
      .timeWindow(Time.minutes(1))
      .apply(new DataWindow())

这是WindowFunction:

class DataWindow extends WindowFunction[Data.Fingerprint, String, String, TimeWindow]  {

    override def apply(key: String,
                       window: TimeWindow,
                       input: Iterable[Fingerprint],
                       out: Collector[String]) {

      out.collect("helo")
    }
  }

推荐答案

我认为问题出在 WindowFunction 的第三个类型参数，即键的类型.因为您在 keyBy 方法中将键声明为 String (keyBy("hash"))，所以无法在编译时确定键的类型.有两种方法可以解决此问题:

I think the problem is with the third type parameter of the WindowFunction, i.e., the type of the key. Because you declare the key in the keyBy method as String (keyBy("hash")), it is not possible to determine the type of the key at compile time. There are two options to resolve this:

1. 使用 keyBy 中的 KeySelector 函数来提取密钥(类似于 keyBy(x: FingerPrint => x.hash)).KeySelector 函数的返回类型在编译时已知，因此您可以使用类型化的 WindowFunction.
2. 将WindowFunction的第三个类型参数的类型改为Tuple.Tuple 是 keyBy 提取的密钥的通用持有者.在您的情况下，它将是 Tuple1 并且可以通过 Tuple1.f0 访问哈希字符串.

1. Use a KeySelector function in keyBy to extract the key (something like keyBy(x: FingerPrint => x.hash)). The return type of the KeySelector function is known at compile time such that you can use a typed WindowFunction.
2. Change the type of the third type parameter of the WindowFunction to Tuple. Tuple is a generic holder for the keys extracted by keyBy. In your case it will be a Tuple1 and the hash String can be accessed by Tuple1.f0.

这篇关于无法使用 WindowStream.apply() 函数应用 WindowFunction的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！