在WindowedStream中查找计数 - 快速

Question

我在Streams的世界里很新，我在第一次尝试时遇到了一些问题。

更具体地说，我试图使用Flink在滑动窗口中实现count和group By功能。

我在正常的DateStream中完成了它，但我无法使它在WindowedStream中工作。

你有什么建议我怎么办？

val parsedStream: DataStream[(String, Response)] = stream
      .mapWith(_.decodeOption[Response])
      .filter(_.isDefined)
      .map { record =>
        (
          s"${record.get.group.group_country}, ${record.get.group.group_state}, ${record.get.group.group_city}",
          record.get
        )
      }

val result: DataStream[((String, Response), Int)] = parsedStream
      .map((_, 1))
      .keyBy(_._1._1)
      .sum(1)

// The output of result is 
// ((us, GA, Atlanta,Response()), 14)
// ((us, SA, Atlanta,Response()), 4)

result
      .keyBy(_._1._1)
      .timeWindow(Time.seconds(5))

//the following part doesn't compile

      .apply(
        new WindowFunction[(String, Int), (String, Int), String, TimeWindow] {
          def apply(
                   key: Tuple,
                   window: TimeWindow,
                   values: Iterable[(String, Response)],
                   out: Collector[(String, Int)]
                   ) {}
        }
      )

编译错误：

overloaded method value apply with alternatives:
  [R](function: (String, org.apache.flink.streaming.api.windowing.windows.TimeWindow, Iterable[((String, com.flink.Response), Int)], org.apache.flink.util.Collector[R]) => Unit)(implicit evidence$28: org.apache.flink.api.common.typeinfo.TypeInformation[R])org.apache.flink.streaming.api.scala.DataStream[R] <and>
  [R](function: org.apache.flink.streaming.api.scala.function.WindowFunction[((String, com.flink.Response), Int),R,String,org.apache.flink.streaming.api.windowing.windows.TimeWindow])(implicit evidence$27: org.apache.flink.api.common.typeinfo.TypeInformation[R])org.apache.flink.streaming.api.scala.DataStream[R]
 cannot be applied to (org.apache.flink.streaming.api.functions.windowing.WindowFunction[((String, com.flink.Response), Int),(String, com.flink.Response),String,org.apache.flink.streaming.api.windowing.windows.TimeWindow]{def apply(key: String,window: org.apache.flink.streaming.api.windowing.windows.TimeWindow,input: Iterable[((String, com.flink.Response), Int)],out: org.apache.flink.util.Collector[(String, com.flink.Response)]): Unit})
      .apply(

Answer 1

这是一个我们可以处理的更简单的例子

val source: DataStream[(JsonField, Int)] = env.fromElements(("hello", 1), ("hello", 2))

    val window2 = source
      .keyBy(0)
      .timeWindow(Time.minutes(1))
      .apply(new WindowFunction[(JsonField, Int), Int, String, TimeWindow] {})

Answer 2

我已经尝试了你的代码并发现了错误，在你为WindowFunction声明类型时似乎有错误。

文档说，WindowFunction的预期类型是WindowFunction[IN, OUT, KEY, W <: Window]。现在，如果您查看您的代码，您的IN是您正在计算窗口的数据流的类型。流的类型是((String, Response), Int)，而不是代码(String, Int)中声明的。

如果您要将未编译的零件更改为：

.apply(new WindowFunction[((String, Response), Int), (String, Response), String, TimeWindow] {
        override def apply(key: String, window: TimeWindow, input: Iterable[((String, Response), Int)], out: Collector[(String, Response)]): Unit = ???
})

编辑：至于第二个例子，由于同样的原因，错误发生。当你使用keyBy和Tuple时你有两个可能的函数来使用keyBy(fields: Int*)，它使用整数来访问使用索引提供的元组字段（这就是你所使用的）。还有keyBy(fun: T => K)，你提供了一个函数来提取将要使用的密钥。

但是这些函数之间有一个重要的区别，其中一个函数返回密钥为JavaTuple，另一个函数返回密钥及其确切类型。所以基本上如果你在简化的例子中将String更改为Tuple它应该清楚地编译。