其他分享
首页 > 其他分享> > Flink-富函数(Rich Functions)

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 版本。它与常规函数的不同在于,可以获取运行环境的上下文,并拥有一些生命周期方法,所以可以实现更复杂的功能。 Rich Function 有一个生命周期的概念。典型的生命周期方法有:
// 富函数,可以获取到运行时上下文,还有一些生命周期
  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调用完之后。一般做收尾工作,比如关闭连接,或者清空状态
    }
  }

 

     

标签:Functions,函数,Flink,value,filter,SensorReading,Rich,override,String
来源: https://www.cnblogs.com/EnzoDin/p/16357518.html