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