Flink-富函数(Rich Functions)

4.富函数（Rich Functions）

4.1 函数类（Function Classes）

Flink 暴露了所有 udf 函数的接口(实现方式为接口或者抽象类)。例如MapFunction, FilterFunction, ProcessFunction 等等。

下面例子实现了 FilterFunction 接口：

class FilterFilter extends FilterFunction[String] {
    override def filter(value: String): Boolean = {
        value.contains("flink")
    }
}
val flinkTweets = tweets.filter(new FlinkFilter)      

还可以将函数实现成匿名类

val flinkTweets = tweets.filter(
    new RichFilterFunction[String] {
        override def filter(value: String): Boolean = {
            value.contains("flink")
        }
    }
)      

我们 filter 的字符串"flink"还可以当作参数传进去。

class MyFilter(keyWord: String) extends FilterFunction[SensorReading]{
    override def filter(value: SensorReading): Boolean = {
      value.id.contains(keyWord)
    }
  }
 
val dataStream: DataStream[SensorReading] = inputStream
.map(
   data => {
     val arr = data.split(",")
     SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
   }
)      

4.2 匿名函数（Lambda Functions）

val tweets: DataStream[String] = ...
val flinkTweets = tweets.filter(_.contains("flink"))      

4.3 富函数（Rich Functions）

“富函数”是 DataStream API 提供的一个函数类的接口，所有 Flink 函数类都有其 Rich 版本。它与常规函数的不同在于，可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

• RichMapFunction
• RichFlatMapFunction
• RichFilterFunction
• ...

• open()方法是 rich function 的初始化方法，当一个算子例如 map 或者 filter被调用之前 open()会被调用。
• close()方法是生命周期中的最后一个调用的方法，做一些清理工作。
• getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息，例如函数执行的并行度，任务的名字，以及 state 状态

// 富函数，可以获取到运行时上下文，还有一些生命周期
  class MyRichMap extends RichMapFunction[SensorReading, String]{

    override def open(parameters: Configuration): Unit = {
      //做一些初始化操作。比如map方法需要交互数据库，数据库连接可以在open里边做
      //getRuntimeContext()
    }

    override def map(value: SensorReading): String = {
      value.id + " temperature"
    }

    override def close(): Unit = {
      //map调用完之后。一般做收尾工作，比如关闭连接，或者清空状态
    }
  }