无法使用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屋！