Apache Flink:ProcessWindowFunction实现 [英] Apache Flink: ProcessWindowFunction implementation

问题描述

我正在尝试使用Scala在我的Apache Flink项目中使用ProcessWindowFunction.不幸的是，我已经无法实现像Apache Flink文档中所使用的那样基本的ProcessWindowFunction.

I am trying to use a ProcessWindowFunction in my Apache Flink project using Scala. Unfortunately, I already fail at implementing a basic ProcessWindowFunction like it is used in the Apache Flink Documentation.

这是我的代码:

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.fiware.cosmos.orion.flink.connector.{NgsiEvent, OrionSource}
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.util.Collector
import scala.collection.TraversableOnce

object StreamingJob {
 def main(args: Array[String]) {

 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val eventStream = env.addSource(new OrionSource(9001))

 val processedDataStream = eventStream.flatMap(event => event.entities)
   .map(entity => (entity.id, entity.attrs("temperature").value.asInstanceOf[String]))
     .keyBy(_._1)
     .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
     .process(new MyProcessWindowFunction())

 env.execute("Socket Window NgsiEvent")
 }
}


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

def process(key: String, context: Context, input: Iterable[(String, String)], out: Collector[String]): Unit = {
  var count: Int = 0
  for (in <- input) {
    count = count + 1
  }
  out.collect(s"Window ${context.window} count: $count")
 }
}

从IntelliJ中，我得到以下提示:

From IntelliJ I get the following hints:

1)显示创建新类对象的位置:

1) This is shown where the new class object is created:

Type mismatch, expected: ProcessWindowFunction[(String, String), NotInferedR, String, TimeWindow], actual: MyProcessWindowFunction

2)这在班上直接显示:

2) This is shown directly at the class:

Class 'MyProcessWindowFunction' must either be declared abstract or implement abstract member 'process(key:KEY, context:ProcessWindowFunction.Context, iterable:Iterable<IN>, collector:Collector<OUT>):void' in 'org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction'

构建代码会向我显示以下错误:

Building the code shows me the following error:

Error:(51, 16) type mismatch;
found   : org.apache.flink.MyProcessWindowFunction
required: 
org.apache.flink.streaming.api.scala.function.ProcessWindowFunction[(String, String),?,String,org.apache.flink.streaming.api.windowing.windows.TimeWindow]
  .process(new MyProcessWindowFunction())

我非常感谢您的帮助.

推荐答案

花了一些时间与另外2个人进行调试之后，我们终于设法找到了问题.

After spending some time debugging with 2 more people we finally managed to find the problem.

在我的代码中，我使用了以下导入:

In my code I used the following import:

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction

但是使用Scala时正确的导入似乎是:

But the correct import when using Scala seems to be:

import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction

这篇关于Apache Flink:ProcessWindowFunction实现的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！