Flink流-在windows中应用函数

Question

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val inputStream = env.readTextFile("dataset.txt")
      .map(transformStream(_))
      .assignAscendingTimestamps(_.eventTime)
      .keyBy(_.id)
      .timeWindow(Time.seconds(windowSize),Time.seconds(slidingStep))

def transformStream(input: String): EventStream = {...}

case class EventStream(val eventTime: Long, val id: String, actualEvent: String)

我想做的是对每个窗口批处理的每个分区应用一个通用函数，也许应用一个复杂的处理算法或类似的东西。我已经在DataStream API中看到了该方法的应用，但我不理解它是如何工作的。在Flink API中,它说它的用法与Scala中的用法相同：

inputStream.apply { WindowFunction }

有人能解释一下apply方法是做什么的或者是如何使用的吗？Scala中的示例会更好。apply方法是否符合我的要求？

Answer 1

所以基本上有两个可能的方向，根据你想要做的计算类型。可以使用:fold/reduce/aggrege或更多您已经提到的泛型-apply。所有这些应用于windows的密钥。

对于apply，它是应用计算的一种非常通用的方法。最基本的版本（在Scala中）是：

def apply[R: TypeInformation](function: (K, W, Iterable[T],Collector[R]) => Unit): DataStream[R] 

其中函数采用4个参数：

null

val stream: DataStream[(String,Int)] =   ...

stream.keyBy(_._1)
      .window(EventTimeSessionWindows.withGap(Time.seconds(conf.sessionGap())))
      .apply((e1, e2) => (e1._1, e1._2 + e2._2),
             (key, window, in, out: Collector[(String, Long, Long, Int)]) => {
                out.collect((key, window.getStart, window.getEnd, in.map(_._2).sum))
      })