Flink-转换算子

基本转换算子

map(映射)

filter（过滤）

flatMap（扁平映射）

聚合算子

keyBy（按键分区）

简单聚合

reduce（归约聚合）

UDF介绍

函数类

富函数类

数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个 DataStream 转换为新的 DataStream。一个 Flink 程序的核心，其实就是所有的转换操作，它们决定了处理的业务逻辑。

基本转换算子

map(映射)

主要用于将数据流中的数据进行转换，形成新的数据流。

我们只需要基于 DataStrema 调用 map()方法就可以进行转换处理。方法需要传入的参数是接口 MapFunction 的实现；返回值类型还是 DataStream，不过泛型（流中的元素类型）可能改变。

case class Event(user: String, url: String, timestamp: Long)

object test {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))//转换需求：提取用户名//实现方式1：使用匿名函数stream.map( _.user).print("方式1：")//实现方式2：实现MapFunction接口stream.map(new UserEX).print("方式2：")//执行env.execute()}//定义实现接口类class UserEX extends MapFunction[Event,String]{override def map(t: Event): String = t.user}
}

filter（过滤）

filter()转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉

进行 filter()转换之后的新数据流的数据类型与原数据流是相同的。filter()转换需要传入的参数需要实现 FilterFunction 接口，而 FilterFunction 内要实现 filter()方法，就相当于一个返回布尔类型的条件表达式。

case class Event(user: String, url: String, timestamp: Long)

object f5 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))//过滤出用户名为王五的点击事件//方式1：使用匿名函数stream.filter(_.user == "王五").print("方式1：")//方式2：使用匿名函数stream.filter(new UserF).print("方式2：")//执行env.execute()}//定义实现接口类class UserF extends FilterFunction[Event] {override def filter(t: Event): Boolean = t.user == "王五"}
}

flatMap（扁平映射）

flatMap()操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap()可以认为是“扁平化” （flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。

同 map()一样，flatMap()也可以使用 Lambda 表达式或者 FlatMapFunction 接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流的数据类型相同，也可以不同。

case class Event(user: String, url: String, timestamp: Long)

object f5 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))//测试灵活的输出形式:如果点击用户为李四 则输出李四stream.flatMap(new UserF).print()//执行env.execute()}//定义实现接口类class UserF extends FlatMapFunction[Event,String] {override def flatMap(t: Event, collector: Collector[String]): Unit = {//如果当前数据是Mary的点击事件则直接输出userif(t.user == "李四"){collector.collect(t.user)}}}
}

聚合算子

keyBy（按键分区）

对于 Flink 而言，DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区；这个操作就是通过 keyBy()来完成的。

keyBy()是聚合前必须要用到的一个算子。keyBy()通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slots）。

基于不同的 key，流中的数据将被分配到不同的分区中去，如图 5-8 所示；这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot 中进行处理了。

case class Event(user: String, url: String, timestamp: Long)

object f6 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("李四", "04", 2000L),Event("王五", "01", 6000L),Event("赵六", "03", 1000L))//按键分区//1、接口实现stream.keyBy( new MyKeyb ).print()//匿名函数实现stream.keyBy( k => k.user)//执行env.execute()}//定义实现接口类class MyKeyb extends  KeySelector[Event,String]{override def getKey(in: Event): String = in.user}
}

简单聚合

有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink 为我们内置实现了一些最基本、最简单的聚合 API，主要有以下几种：

sum()：在输入流上，对指定的字段做叠加求和的操作。
min()：在输入流上，对指定的字段求最小值。
max()：在输入流上，对指定的字段求最大值。
minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包含字段最小值的整条数据。
maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与 min()/minBy()完全一致。

object f6 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("张三", "01", 6000L),Event("李四", "04", 2000L),Event("李四", "05", 6000L),Event("王五", "01", 1000L),Event("赵六", "03", 1000L))//使用 user 作为分组的字段，并计算最大的时间戳stream.keyBy( new MyKeyb ).maxBy("timestamp").print()//执行env.execute()}//定义实现接口类class MyKeyb extends  KeySelector[Event,String]{override def getKey(in: Event): String = in.user}
}

reduce（归约聚合）

与简单聚合类似，reduce()操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。

调用 KeyedStream 的 reduce()方法时，需要传入一个参数，实现 ReduceFunction 接口。接口在源码中的定义如下：

case class Event(user: String, url: String, timestamp: Long)

object f6 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("张三", "01", 6000L),Event("李四", "04", 2000L),Event("李四", "05", 6000L),Event("王五", "01", 1000L),Event("赵六", "03", 1000L))//归约聚合：统计用户点击页面的数量，提取当前最活跃的用户stream.map( d => (d.user,1L)).keyBy(_._1).reduce( new MyS) //统计每个用户的活跃度.keyBy( d => true) //将所有的数据按照同样的key分到同一个组中.reduce((a,b) => if(b._2 > a._2) b else a) //选取当前最活跃的用户.print() //输出结果//执行env.execute()}//定义实现接口类class MyS extends ReduceFunction[(String,Long)]{override def reduce(t: (String, Long), t1: (String, Long)): (String, Long) = (t._1,t._2+t1._2)}
}

UDF介绍

函数类

对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类，例如 MapFunction、FilterFunction、ReduceFunction 等。所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。

下面例子实现了 FilterFunction 接口，用来筛选 url 中包含“01”的内容：

object f7 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("张三", "01", 6000L),Event("李四", "04", 2000L),Event("李四", "05", 6000L),Event("王五", "01", 1000L),Event("赵六", "03", 1000L))//测试自定义函数类UDF的用法：筛选url中包含关键字“01”的事件//1、实现一个自定义的函数类stream.filter( new MyFFun).print()//2、使用匿名类 筛选url中包含关键字“04”的事件stream.filter( new FilterFunction[Event] {override def filter(t: Event): Boolean = t.url.contains("04")}).print()//3、使用匿名函数stream.filter(d => d.url.contains("05"))//执行env.execute()}//定义实现接口类class MyFFun extends FilterFunction[Event]{override def filter(t: Event): Boolean = t.url.contains("01")}
}

富函数类

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其 Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、 RichReduceFunction 等。

与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能,典型的生命周期方法有：

open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调用。所以像文件 IO 流的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在 open()方法中完成。
close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。

object f8 {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//设置全局并行度env.setParallelism(1)//创建当前样例类Event的对象val stream: DataStream[Event] = env.fromElements(Event("张三", "01", 1000L),Event("张三", "01", 6000L),Event("李四", "04", 2000L),Event("李四", "05", 6000L),Event("王五", "01", 1000L),Event("赵六", "03", 1000L))//自定义一个RichMapFunction 测试富函数类的功能stream.map(new MyRichMap).print()//执行env.execute()}//定义实现接口类class MyRichMap extends RichMapFunction[Event,Long]{override def open(parameters: Configuration): Unit = {println("索引号为："+ getRuntimeContext.getIndexOfThisSubtask + "的任务开始")}override def map(in: Event): Long = in.timestamp //输出一个长整型的时间戳override def close(): Unit = {println("索引号为："+ getRuntimeContext.getIndexOfThisSubtask + "的任务结束")}}
}

适合使用的场景

class MyFlatMap extends RichFlatMapFunction[IN,OUT]{override def open(parameters: Configuration): Unit = {// 做一些初始化工作// 例如建立一个和 MySQL 的连接}override def flatMap(value: IN, out: Collector[OUT]): Unit = {// 对数据库进行读写}override def close(): Unit = {// 清理工作，关闭和 MySQL 数据库的连接。}
}