CODEWITHSUNDEEP
scalastreamstreamingapache-flinkflink-streaming
pik4
pik4

Reputation: 1373

Find Top K elements in WindowedStream - Flink

I am pretty new in the world of Streams and I am facing some issues in my first try.

What I would like to do is to find the Top K element in the window: WindowdStream below. I tried to implement my own function but not sure how it actually works.

Seems that doesn't print anything

May you have any hint?

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)

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])

Upvotes: 1

Views: 943

Answers (1)

David Anderson
David Anderson

Reputation: 43499

When your Flink DataStream job fails to produce any output, the usual suspects are:

  • the job doesn't call execute() on the StreamExecutionEnvironment (e.g., env.execute())
  • the job doesn't have a sink attached (e.g., TopLocations.print())
  • the job is meant to use event time, but the Watermarks aren't set up correctly or an idle source is preventing the watermarks from advancing
  • the job is writing to the taskmanager logs, but no one noticed
  • the serializer for the output type produces no output

Without more information it's difficult to guess which of these might be the problem in this case.

Upvotes: 1

Related Questions