在windowedstream中查找前k个元素-flink

oxf4rvwz  于 2021-06-21  发布在  Flink
关注(0)|答案(1)|浏览(348)

我对溪流的世界还很陌生,我在第一次尝试时就面临着一些问题。
我想做的是在 window: WindowdStream 在下面。我试图实现我自己的功能,但不知道它实际上是如何工作的。
好像什么都没印出来
你有什么提示吗?

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

val topLocations = parsedStream
      .keyBy(_._1)
      .timeWindow(Time.days(7))
      .process(new SortByCountFunction)

排序计数函数

class SortByCountFunction
    extends ProcessWindowFunction[(String, Response), MeetUpLocationWindow, String, TimeWindow] {

    override def process(key: String,
                         context: Context,
                         elements: Iterable[(String, Response)],
                         out: Collector[MeetUpLocationWindow]): Unit = {

      val count: Map[String, Iterable[(String, Response)]] = elements.groupBy(_._1)

      val locAndCount: Seq[MeetUpLocation] = count.toList.map(tmp => {
        val location: String = tmp._1
        val meetUpList: Iterable[(String, Response)] = tmp._2
        MeetUpLocation(location, tmp._2.size, meetUpList.map(_._2).toList)
      })

      val output: List[MeetUpLocation] = locAndCount.sortBy(tup => tup.count).take(20).toList

      val windowEnd = context.window.getEnd

      out.collect(MeetUpLocationWindow(windowEnd, output))
    }
  }

case class MeetUpLocationWindow(endTs: Long, locations: List[MeetUpLocation])

case class MeetUpLocation(location: String, count: Int, meetUps: List[Response])
epfja78i

epfja78i1#

当flink datastream作业无法生成任何输出时,通常的怀疑是:
作业不会在streamexecutionenvironment上调用execute()(例如。, env.execute() )
作业没有连接Flume(例如。, TopLocations.print() )
该作业旨在使用事件时间,但水印设置不正确,或者空闲源阻止水印前进
作业正在写入taskmanager日志,但没有人注意到
输出类型的序列化程序不生成输出
如果没有更多的信息,就很难猜测这些问题中的哪一个可能是本例中的问题。

相关问题