Collector in Flink. What does it do?

user_2728349 0
Viewed 1583

I'm learning Flink and one of the things is confusing for me is the fact of using an object called Collector. For example in the flatmap function. What's this Collector and its method collect? and why for example a map function doesn't need to pass results by explicitly using it ?

here there can be seen some examples of using Collector in the flatmap function: https://www.programcreek.com/scala/org.apache.flink.util.Collector

also, if I search for where the Collector would be placed in Flink Architecture I don't find any diagram with that mapping

apache-flink scala
2 Answers

Flink passes a Collector to any user function that has the possibility of emitting an arbitrary number of stream elements. A map function doesn’t use a Collector because it performs a one-to-one transformation, with the return value of the map function being the output. Whereas a flatmap can emit zero, one, or many stream elements for each event, which makes the Collector a convenient way to accommodate this.

user_20008230

as you know, if you want one piece to produce N outputs in the data stream, you can use Collector to encapsulate the output data in flatmap,on the contrary,Map usually produces one-to-one data, so doesn't need to use it.In a sense,Collector has a wide range of internal applications. You can take a look at org.apache.flink.streaming.api.operators.Output(extend from Collector) \org.apache.flink.runtime.operators.shipping.OutputCollector ,they are usually used to collect records and emits them to writer.and so on,collect be called when needs to write data.

Examples (not necessarily accurate)：

There are three definitions of Scala source code for flatMap. Let's take a look at the definition of the first one.

  /**
   * Creates a new DataStream by applying the given function to every element and flattening
   * the results.
   */
  def flatMap[R: TypeInformation](fun: (T, Collector[R]) => Unit): DataStream[R] = {
    if (fun == null) {
      throw new NullPointerException("FlatMap function must not be null.")
    }
    val cleanFun = clean(fun)
    val flatMapper = new FlatMapFunction[T, R] {
      def flatMap(in: T, out: Collector[R]) { cleanFun(in, out) }
    }
    flatMap(flatMapper)
  }

Examples of using this method are as follows：

text.flatMap((input: String, out: Collector[String]) => {
  input.split(" ").foreach(out.collect)
})

In this method, we need to send the data manually through Collector

Then let's take a look at the second definition in the source code:

  /**
   * Creates a new DataStream by applying the given function to every element and flattening
   * the results.
   */
  def flatMap[R: TypeInformation](fun: T => TraversableOnce[R]): DataStream[R] = {
    if (fun == null) {
      throw new NullPointerException("FlatMap function must not be null.")
    }
    val cleanFun = clean(fun)
    val flatMapper = new FlatMapFunction[T, R] {
      def flatMap(in: T, out: Collector[R]) { cleanFun(in) foreach out.collect }
    }
    flatMap(flatMapper)
  }

Instead of using Collector to collect the output, here we output a list directly, and Flink helps us flatten the list. Using TraversableOnce also causes us to return a list anyway, even if it is an empty list, otherwise we cannot match the definition of the function.

text.flatMap(input => {
  if (input.size > 15) {
    input.split(" ")
  } else {
    Seq.empty
  }
})

You can find a lot of similar places, as long as it is about sending data records, you can almost see Collector.

user_63852050
Related