标签：Functions 函数 Flink value filter SensorReading Rich override String

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
•  ...

Rich Function 有一个生命周期的概念。典型的生命周期方法有：

• 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调用完之后。一般做收尾工作，比如关闭连接，或者清空状态
    }
  }

 

     

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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。