直入正题！
Flink 和 Spark 类似，也是一种一站式处理的框架；既可以进行批处理（DataSet），也可以进行实时处理（DataStream）。
所以下面将 Flink 的算子分为两大类：一类是 DataSet，一类是 DataStream。
DataSet
一、Source 算子
1. fromCollection
fromCollection：从本地集合读取数据
例：
val env = ExecutionEnvironment.getExecutionEnvironmentval textDataSet: DataSet[String] = env.fromCollection( List("1,张三", "2,李四", "3,王五", "4,赵六"))
复制代码
2. readTextFile
readTextFile：从文件中读取
val textDataSet: DataSet[String] = env.readTextFile("/data/a.txt")
复制代码
3. readTextFile：遍历目录
readTextFile 可以对一个文件目录内的所有文件，包括所有子目录中的所有文件的遍历访问方式
val parameters = new Configuration// recursive.file.enumeration 开启递归parameters.setBoolean("recursive.file.enumeration", true)val file = env.readTextFile("/data").withParameters(parameters)
复制代码
4. readTextFile：读取压缩文件
对于以下压缩类型，不需要指定任何额外的 inputformat 方法，flink 可以自动识别并且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。
| 压缩方法 | 文件扩展名 | 是否可并行读取 |
| :---------: | :--: | :-----------: |
| DEFLATE | .deflate | no |
| GZip | .gz .gzip | no |
| Bzip2 | .bz2 | no |
| XZ | .xz | no |
val file = env.readTextFile("/data/file.gz")
复制代码
二、Transform 转换算子
因为 Transform 算子基于 Source 算子操作，所以首先构建 Flink 执行环境及 Source 算子，后续 Transform 算子操作基于此：
val env = ExecutionEnvironment.getExecutionEnvironmentval textDataSet: DataSet[String] = env.fromCollection( List("张三,1", "李四,2", "王五,3", "张三,4"))
复制代码
1. map
将 DataSet 中的每一个元素转换为另外一个元素
// 使用map将List转换为一个Scala的样例类
case class User(name: String, id: String)
val userDataSet: DataSet[User] = textDataSet.map { text => val fieldArr = text.split(",") User(fieldArr(0), fieldArr(1))}userDataSet.print()
复制代码
2. flatMap
将 DataSet 中的每一个元素转换为 0...n 个元素。
// 使用flatMap操作，将集合中的数据：// 根据第一个元素，进行分组// 根据第二个元素，进行聚合求值
val result = textDataSet.flatMap(line => line) .groupBy(0) // 根据第一个元素，进行分组 .sum(1) // 根据第二个元素，进行聚合求值 result.print()
复制代码
3. mapPartition
将一个分区中的元素转换为另一个元素
// 使用mapPartition操作，将List转换为一个scala的样例类
case class User(name: String, id: String)
val result: DataSet[User] = textDataSet.mapPartition(line => { line.map(index => User(index._1, index._2)) }) result.print()
复制代码
4. filter
过滤出来一些符合条件的元素，返回 boolean 值为 true 的元素
val source: DataSet[String] = env.fromElements("java", "scala", "java")val filter:DataSet[String] = source.filter(line => line.contains("java"))//过滤出带java的数据filter.print()
复制代码
5. reduce
可以对一个 dataset 或者一个 group 来进行聚合计算，最终聚合成一个元素
// 使用 fromElements 构建数据源val source = env.fromElements(("java", 1), ("scala", 1), ("java", 1))// 使用map转换成DataSet元组val mapData: DataSet[(String, Int)] = source.map(line => line)// 根据首个元素分组val groupData = mapData.groupBy(_._1)// 使用reduce聚合val reduceData = groupData.reduce((x, y) => (x._1, x._2 + y._2))// 打印测试reduceData.print()
复制代码
6. reduceGroup
将一个 dataset 或者一个 group 聚合成一个或多个元素。
reduceGroup 是 reduce 的一种优化方案；
它会先分组 reduce，然后在做整体的 reduce；这样做的好处就是可以减少网络 IO
// 使用 fromElements 构建数据源val source: DataSet[(String, Int)] = env.fromElements(("java", 1), ("scala", 1), ("java", 1))// 根据首个元素分组val groupData = source.groupBy(_._1)// 使用reduceGroup聚合val result: DataSet[(String, Int)] = groupData.reduceGroup { (in: Iterator[(String, Int)], out: Collector[(String, Int)]) => val tuple = in.reduce((x, y) => (x._1, x._2 + y._2)) out.collect(tuple) }// 打印测试result.print()
复制代码
7. minBy 和 maxBy
选择具有最小值或最大值的元素
// 使用minBy操作，求List中每个人的最小值// List("张三,1", "李四,2", "王五,3", "张三,4")
case class User(name: String, id: String)// 将List转换为一个scala的样例类val text: DataSet[User] = textDataSet.mapPartition(line => { line.map(index => User(index._1, index._2)) }) val result = text .groupBy(0) // 按照姓名分组 .minBy(1) // 每个人的最小值
复制代码
8. Aggregate
在数据集上进行聚合求最值（最大值、最小值）
val data = new mutable.MutableList[(Int, String, Double)] data.+=((1, "yuwen", 89.0)) data.+=((2, "shuxue", 92.2)) data.+=((3, "yuwen", 89.99))// 使用 fromElements 构建数据源val input: DataSet[(Int, String, Double)] = env.fromCollection(data)// 使用group执行分组操作val value = input.groupBy(1) // 使用aggregate求最大值元素 .aggregate(Aggregations.MAX, 2) // 打印测试value.print()
复制代码
Aggregate 只能作用于元组上
注意：
要使用 aggregate，只能使用字段索引名或索引名称来进行分组 groupBy(0) ，否则会报一下错误:
Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support
grouping with KeySelector functions, yet.
9. distinct
去除重复的数据
// 数据源使用上一题的// 使用distinct操作，根据科目去除集合中重复的元组数据
val value: DataSet[(Int, String, Double)] = input.distinct(1)value.print()
复制代码
10. first
取前 N 个数
11. join
将两个 DataSet 按照一定条件连接到一起，形成新的 DataSet
// s1 和 s2 数据集格式如下：// DataSet[(Int, String,String, Double)]
val joinData = s1.join(s2) // s1数据集 join s2数据集 .where(0).equalTo(0) { // join的条件 (s1, s2) => (s1._1, s1._2, s2._2, s1._3) }
复制代码
12. leftOuterJoin
左外连接,左边的 Dataset 中的每一个元素，去连接右边的元素
此外还有：
rightOuterJoin：右外连接,左边的 Dataset 中的每一个元素，去连接左边的元素
fullOuterJoin：全外连接,左右两边的元素，全部连接
下面以 leftOuterJoin 进行示例：
val data1 = ListBuffer[Tuple2[Int,String]]() data1.append((1,"zhangsan")) data1.append((2,"lisi")) data1.append((3,"wangwu")) data1.append((4,"zhaoliu"))
val data2 = ListBuffer[Tuple2[Int,String]]() data2.append((1,"beijing")) data2.append((2,"shanghai")) data2.append((4,"guangzhou"))
val text1 = env.fromCollection(data1)val text2 = env.fromCollection(data2)
text1.leftOuterJoin(text2).where(0).equalTo(0).apply((first,second)=>{ if(second==null){ (first._1,first._2,"null") }else{ (first._1,first._2,second._2) } }).print()
复制代码
13. cross
交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集
和 join 类似，但是这种交叉操作会产生笛卡尔积，在数据比较大的时候，是非常消耗内存的操作
val cross = input1.cross(input2){ (input1 , input2) => (input1._1,input1._2,input1._3,input2._2) }
cross.print()
复制代码
14. union
联合操作，创建包含来自该数据集和其他数据集的元素的新数据集,不会去重
val unionData: DataSet[String] = elements1.union(elements2).union(elements3)// 去除重复数据val value = unionData.distinct(line => line)
复制代码
15. rebalance
Flink 也有数据倾斜的时候，比如当前有数据量大概 10 亿条数据需要处理，在处理过程中可能会发生如图所示的状况：
这个时候本来总体数据量只需要 10 分钟解决的问题，出现了数据倾斜，机器 1 上的任务需要 4 个小时才能完成，那么其他 3 台机器执行完毕也要等待机器 1 执行完毕后才算整体将任务完成；
所以在实际的工作中，出现这种情况比较好的解决方案就是接下来要介绍的—rebalance（内部使用 round robin 方法将数据均匀打散。这对于数据倾斜时是很好的选择。）
// 使用rebalance操作，避免数据倾斜val rebalance = filterData.rebalance()
复制代码
16. partitionByHash
按照指定的 key 进行 hash 分区
val data = new mutable.MutableList[(Int, Long, String)]data.+=((1, 1L, "Hi"))data.+=((2, 2L, "Hello"))data.+=((3, 2L, "Hello world"))
val collection = env.fromCollection(data)val unique = collection.partitionByHash(1).mapPartition{ line => line.map(x => (x._1 , x._2 , x._3))}
unique.writeAsText("hashPartition", WriteMode.NO_OVERWRITE)env.execute()
复制代码
17. partitionByRange
根据指定的 key 对数据集进行范围分区
val data = new mutable.MutableList[(Int, Long, String)]data.+=((1, 1L, "Hi"))data.+=((2, 2L, "Hello"))data.+=((3, 2L, "Hello world"))data.+=((4, 3L, "Hello world, how are you?"))
val collection = env.fromCollection(data)val unique = collection.partitionByRange(x => x._1).mapPartition(line => line.map{ x=> (x._1 , x._2 , x._3)})unique.writeAsText("rangePartition", WriteMode.OVERWRITE)env.execute()
复制代码
18. sortPartition
根据指定的字段值进行分区的排序
val data = new mutable.MutableList[(Int, Long, String)] data.+=((1, 1L, "Hi")) data.+=((2, 2L, "Hello")) data.+=((3, 2L, "Hello world")) data.+=((4, 3L, "Hello world, how are you?"))
val ds = env.fromCollection(data) val result = ds .map { x => x }.setParallelism(2) .sortPartition(1, Order.DESCENDING)//第一个参数代表按照哪个字段进行分区 .mapPartition(line => line) .collect()
println(result)
复制代码
三、Sink 算子
1. collect
将数据输出到本地集合
2. writeAsText
将数据输出到文件
Flink 支持多种存储设备上的文件，包括本地文件，hdfs 文件等
Flink 支持多种文件的存储格式，包括 text 文件，CSV 文件等
// 将数据写入本地文件result.writeAsText("/data/a", WriteMode.OVERWRITE)
// 将数据写入HDFSresult.writeAsText("hdfs://node01:9000/data/a", WriteMode.OVERWRITE)
复制代码
DataStream
和 DataSet 一样，DataStream 也包括一系列的 Transformation 操作
一、Source 算子
Flink 可以使用 StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源。
Flink 已经提供了若干实现好了的 source functions，当然我们也可以通过实现 SourceFunction 来自定义非并行的 source 或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source。
Flink 在流处理上的 source 和在批处理上的 source 基本一致。大致有 4 大类：
基于本地集合的 source（Collection-based-source）
基于文件的 source（File-based-source）- 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回
基于网络套接字的 source（Socket-based-source）- 从 socket 读取。元素可以用分隔符切分。
自定义的 source（Custom-source）
下面使用 addSource 将 Kafka 数据写入 Flink 为例：
如果需要外部数据源对接，可使用 addSource，如将 Kafka 数据写入 Flink，
先引入依赖：
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11 --><dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka-0.11_2.11</artifactId> <version>1.10.0</version></dependency>
复制代码
将 Kafka 数据写入 Flink：
val properties = new Properties()properties.setProperty("bootstrap.servers", "localhost:9092")properties.setProperty("group.id", "consumer-group")properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")properties.setProperty("auto.offset.reset", "latest")
val source = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
复制代码
基于网络套接字的：
val source = env.socketTextStream("IP", PORT)
复制代码
二、Transform 转换算子
1. map
将 DataSet 中的每一个元素转换为另外一个元素
dataStream.map { x => x * 2 }
复制代码
2. FlatMap
采用一个数据元并生成零个，一个或多个数据元。将句子分割为单词的 flatmap 函数
dataStream.flatMap { str => str.split(" ") }
复制代码
3. Filter
计算每个数据元的布尔函数，并保存函数返回 true 的数据元。过滤掉零值的过滤器
dataStream.filter { _ != 0 }
复制代码
4. *KeyBy*
逻辑上将流分区为不相交的分区。具有相同 Keys 的所有记录都分配给同一分区。在内部，keyBy（）是使用散列分区实现的。指定键有不同的方法。
此转换返回 KeyedStream，其中包括使用被 Keys 化状态所需的 KeyedStream。
5. Reduce
被 Keys 化数据流上的“滚动”Reduce。将当前数据元与最后一个 Reduce 的值组合并发出新值
keyedStream.reduce { _ + _ }
复制代码
6. *Fold*
具有初始值的被 Keys 化数据流上的“滚动”折叠。将当前数据元与最后折叠的值组合并发出新值
val result: DataStream[String] = keyedStream.fold("start")((str, i) => { str + "-" + i })
// 解释：当上述代码应用于序列（1,2,3,4,5）时，输出结果“start-1”，“start-1-2”，“start-1-2-3”，...
复制代码
7. Aggregations
在被 Keys 化数据流上滚动聚合。min 和 minBy 之间的差异是 min 返回最小值，而 minBy 返回该字段中具有最小值的数据元（max 和 maxBy 相同）。
keyedStream.sum(0);
keyedStream.min(0);
keyedStream.max(0);
keyedStream.minBy(0);
keyedStream.maxBy(0);
复制代码
8. *Window*
可以在已经分区的 KeyedStream 上定义 Windows。Windows 根据某些特征（例如，在最后 5 秒内到达的数据）对每个 Keys 中的数据进行分组。这里不再对窗口进行详解，有关窗口的完整说明，请查看这篇文章：
Flink 中极其重要的 Time 与 Window 详细解析
dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));
复制代码
9. *WindowAll*
Windows 可以在常规 DataStream 上定义。Windows 根据某些特征（例如，在最后 5 秒内到达的数据）对所有流事件进行分组。
注意：在许多情况下，这是非并行转换。所有记录将收集在 windowAll 算子的一个任务中。
dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
复制代码
10. Window Apply
将一般函数应用于整个窗口。
注意：如果您正在使用 windowAll 转换，则需要使用 AllWindowFunction。
下面是一个手动求和窗口数据元的函数
windowedStream.apply { WindowFunction }
allWindowedStream.apply { AllWindowFunction }
复制代码
11. Window Reduce
将函数缩减函数应用于窗口并返回缩小的值
windowedStream.reduce { _ + _ }
复制代码
12. Window Fold
将函数折叠函数应用于窗口并返回折叠值
val result: DataStream[String] = windowedStream.fold("start", (str, i) => { str + "-" + i })
// 上述代码应用于序列（1,2,3,4,5）时，将序列折叠为字符串“start-1-2-3-4-5”
复制代码
13. *Union*
两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次数据元
dataStream.union(otherStream1, otherStream2, ...)
复制代码
14. *Window Join*
在给定 Keys 和公共窗口上连接两个数据流
dataStream.join(otherStream) .where(<key selector>).equalTo(<key selector>) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new JoinFunction () {...})
复制代码
15. Interval Join
在给定的时间间隔内使用公共 Keys 关联两个被 Key 化的数据流的两个数据元 e1 和 e2，以便 e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound
am.intervalJoin(otherKeyedStream) .between(Time.milliseconds(-2), Time.milliseconds(2)) .upperBoundExclusive(true) .lowerBoundExclusive(true) .process(new IntervalJoinFunction() {...})
复制代码
16. Window CoGroup
在给定 Keys 和公共窗口上对两个数据流进行 Cogroup
dataStream.coGroup(otherStream) .where(0).equalTo(1) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new CoGroupFunction () {...})
复制代码
17. *Connect*
“连接”两个保存其类型的数据流。连接允许两个流之间的共享状态
DataStream<Integer> someStream = ... DataStream<String> otherStream = ... ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream)
// ... 代表省略中间操作
复制代码
18. *CoMap，CoFlatMap*
类似于连接数据流上的 map 和 flatMap
connectedStreams.map( (_ : Int) => true, (_ : String) => false)connectedStreams.flatMap( (_ : Int) => true, (_ : String) => false)
复制代码
19. *Split*
根据某些标准将流拆分为两个或更多个流
val split = someDataStream.split( (num: Int) => (num % 2) match { case 0 => List("even") case 1 => List("odd") })
复制代码
20. *Select*
从拆分流中选择一个或多个流
SplitStream<Integer> split;DataStream<Integer> even = split.select("even");DataStream<Integer> odd = split.select("odd");DataStream<Integer> all = split.select("even","odd")
复制代码
三、Sink 算子
支持将数据输出到：
本地文件(参考批处理)
本地集合(参考批处理)
HDFS(参考批处理)
除此之外，还支持：
sink 到 kafka
sink 到 mysql
sink 到 redis
下面以 sink 到 kafka 为例：
val sinkTopic = "test"
//样例类case class Student(id: Int, name: String, addr: String, sex: String)val mapper: ObjectMapper = new ObjectMapper()
//将对象转换成字符串def toJsonString(T: Object): String = { mapper.registerModule(DefaultScalaModule) mapper.writeValueAsString(T)}
def main(args: Array[String]): Unit = { //1.创建流执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.准备数据 val dataStream: DataStream[Student] = env.fromElements( Student(8, "xiaoming", "beijing biejing", "female") ) //将student转换成字符串 val studentStream: DataStream[String] = dataStream.map(student => toJsonString(student) // 这里需要显示SerializerFeature中的某一个，否则会报同时匹配两个方法的错误 ) //studentStream.print() val prop = new Properties() prop.setProperty("bootstrap.servers", "node01:9092")
val myProducer = new FlinkKafkaProducer011[String](sinkTopic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), prop) studentStream.addSink(myProducer) studentStream.print() env.execute("Flink add sink")}
复制代码
--end--
评论