在Flink中,流窗口似乎不起作用？

Question

我试图增强显示流的使用的Flink示例.我的目标是使用窗口功能(请参阅window函数调用).我假设下面的代码输出流的最后3个数字的总和.(由于nc -lk 9999在ubuntu上打开了流)实际上,输出总结了输入的所有数字.切换到时间窗口会产生相同的结果,即不会产生窗口.

那是一个错误吗？(使用的版本:github上的最新版本)

object SocketTextStreamWordCount {
  def main(args: Array[String]) {
    val hostName = args(0)
    val port = args(1).toInt
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // Create streams for names and ages by mapping the inputs to the corresponding objects
    val text = env.socketTextStream(hostName, port)    
    val currentMap = text.flatMap { (x:String) => x.toLowerCase.split("\\W+") }
    .filter { (x:String) => x.nonEmpty }      
    .window(Count.of(3)).every(Time.of(1, TimeUnit.SECONDS))
    //  .window(Time.of(5, TimeUnit.SECONDS)).every(Time.of(1, TimeUnit.SECONDS))
      .map { (x:String) => ("not used; just to have a tuple for the sum", x.toInt) }

    val numberOfItems = currentMap.count
    numberOfItems print
    val counts = currentMap.sum( 1 )
    counts print

    env.execute("Scala SocketTextStreamWordCount Example")
  }
}

Answer 1

问题似乎是隐含的转换WindowedDataStream为DataStream.这个隐式转换调用flatten()了WindowedDataStream.

在您的情况下发生的是代码扩展到这:

val currentMap = text.flatMap { (x:String) => x.toLowerCase.split("\\W+") }
    .filter { (x:String) => x.nonEmpty }      
    .window(Count.of(3)).every(Time.of(1, TimeUnit.SECONDS))
    .flatten()   
    .map { (x:String) => ("not used; just to have a tuple for the sum",x.toInt) }    

什么flatten()与flatMap()集合上的类似.它需要窗口流,可以看作是集合([[a,b,c], [d,e,f]])的集合,并将其转换为元素流:[a,b,c,d,e,f].

这意味着您的计数实际上只对已经窗口化和"去窗口化"的原始流进行操作.这看起来根本就没有窗口.

这是一个问题,我将立即努力解决这个问题.(我是Flink提交者之一.)您可以在此处跟踪进度:https://issues.apache.org/jira/browse/FLINK-2096

使用当前API执行此操作的方法是:

val currentMap = text.flatMap { (x:String) => x.toLowerCase.split("\\W+") }
    .filter { (x:String) => x.nonEmpty }   
    .map { (x:String) => ("not used; just to have a tuple for the sum",x.toInt) }    
    .window(Count.of(3)).every(Time.of(1, TimeUnit.SECONDS))

WindowedDataStream有一个sum()方法,因此不会隐式插入flatten()调用.不幸的是,为此你count()不能WindowedDataStream这样做,你必须手动添加一个1字段到元组并计算这些.