Scala/Flink自定义ProcessWindowFunction类型不匹配问题求助

问题原因 #

1. 方法签名不匹配导致未正确覆写父类方法
   你的process方法中，Context参数的类型写为ProcessWindowFunction[(String, String, Int), CommitSummary, String, TimeWindow]#Context，这种冗余写法会导致编译器无法识别该类型与父类定义的Context一致，进而判定你未正确覆写ProcessWindowFunction的核心方法，最终引发类型不匹配错误。
2. 编译器类型推断受阻
   由于方法签名不匹配，IDE无法正确关联MyProcessWindowFunction的泛型参数与process算子要求的参数类型，因此提示“所需类型为ProcessWindowFunction[(String, String, Int), NotInferredR, String, TimeWindow]，实际传入的是MyProcessWindowFunction”。

解决方案 #

只需修改MyProcessWindowFunction中process方法的Context参数类型，简化为父类的Context类型即可——当前类已继承对应泛型版本的ProcessWindowFunction，直接使用Context就能正确匹配父类方法签名。

修改后的完整MyProcessWindowFunction代码：

import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
import org.apache.flink.util.Collector
import util.Protocol.CommitSummary

import java.lang
import java.text.SimpleDateFormat
import scala.collection.JavaConverters._

class MyProcessWindowFunction extends ProcessWindowFunction[
  (String, String, Int),
  CommitSummary,
  String,
  TimeWindow
] {

  override def process(
    key: String,
    context: Context, // 简化为父类的Context类型
    iterable: lang.Iterable[(String, String, Int)],
    collector: Collector[CommitSummary]
  ): Unit = {
    val elements = iterable.asScala
    val simpleDateFormat = new SimpleDateFormat("dd-MM-yyyy")
    val windowStart = context.window.getStart
    val date = simpleDateFormat.format(windowStart)

    val amountOfCommits = elements.size
    val committerCounts = elements.groupBy(_._2).mapValues(_.size)
    val amountOfCommitters = committerCounts.size
    val totalChanges = elements.map(_._3).sum

    val maxCommits = committerCounts.values.max
    val topCommitters = committerCounts.filter(_._2 == maxCommits).keys.toList.sorted.mkString(",")

    val commitSummary = CommitSummary(key, date, amountOfCommits, amountOfCommitters, totalChanges, topCommitters)
    collector.collect(commitSummary)
  }
}

额外说明 #

• 移除了不必要的DataStream导入，精简代码结构
• 若仍存在类型推断问题，可在调用process时显式指定泛型参数（一般无需此操作）：

  .process[(String, String, Int), CommitSummary, String, TimeWindow](new MyProcessWindowFunction)
  

内容的提问来源于stack exchange，提问作者StunnerDH