一、Flink 4种不同层次的API
Flink中提供了4种不同层次的API,每种API在简洁和易表达之间有自己的权衡,适用于不同的场景。目前上面3个会用得比较多。
• 低级API(Stateful Stream Processing):提供了对时间和状态的细粒度控制,简洁性和易用性较差,主要应用在一些复杂事件处理逻辑上。
• 核心API(DataStream/DataSet API):主要提供了针对流数据和批数据的处理,是对低级API进行了一些封装,提供了filter、sum、max、min等高级函数,简单易用,所以这些API在工作中应用还是比较
广泛的。
• Table API:一般与DataSet或者DataStream紧密关联,可以通过一个DataSet或DataStream创建出一个Table,然后再使用类似于filter, join,或者 select这种操作。最后还可以将一个Table对象转成DataSet或DataStream。
• SQL:Flink的SQL底层是基于Apache Calcite,Apache Calcite实现了标准的SQL,使用起来比其他API更加灵活,因为可以直接使用SQL语句。Table API和SQL可以很容易地结合在一块使用,因为它们都返回Table对象。
针对这些API我们主要学习下面这些
二、DataStream API
DataStream API主要分为3块:DataSource、Transformation、DataSink。
- DataSource是程序的输入数据源。
- Transformation是具体的操作,它对一个或多个输入数据源进行计算处理,例如map、flatMap和filter等操作。
- DataSink是程序的输出,它可以把Transformation处理之后的数据输出到指定的存储介质中。
(一)DataStream API之DataSoure
DataSource是程序的输入数据源,Flink提供了大量内置的DataSource,也支持自定义DataSource,不过目前Flink提供的这些已经足够我们正常使用了。
Flink提供的内置输入数据源:包括基于socket、基于Collection,还有就是Flink还提供了一批Connectors,可以实现读取第三方数据源,
说明:
(1)Flink 内置:表示Flink中默认自带的。
(2)Apache Bahir:表示需要添加这个依赖包之后才能使用的。
(3)针对source的这些Connector,我们在实际工作中最常用的就是Kafka
(4)当程序出现错误的时候,Flink的容错机制能恢复并继续运行程序,这种错误包括机器故障、网络故障、程序故障等
(5)针对Flink提供的常用数据源接口,如果程序开启了checkpoint快照机制,Flink可以提供这些容错性保证
1. 使用集合创建DataStream(scala代码)
package com.imooc.scala.streamimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}object StreamCollectionSourceScala {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._//使用collection集合生成DataStreamval text: DataStream[Int] = env.fromCollection(Array(1, 2, 3, 4, 5))text.print().setParallelism(1)env.execute("StreamCollectionSourceScala")}
}
2. 使用集合创建DataStream (java代码)
package com.imooc.java.stream;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Arrays;public class StreamCollectionSourceJava {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//使用Collection集合生成DataStreamDataStreamSource<Integer> text = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5));text.print().setParallelism(1);env.execute("StreamCollectionSourceJava");}
}
(二)、DataStream API之Transformation
transformation是Flink程序的计算算子,负责对数据进行处理,Flink提供了大量的算子,其实Flink中的大部分算子的使用和spark中算子的使用是一样的,下面我们来看一下:
1. union
- 表示合并多个流,但是多个流的数据类型必须一致,多个流join之后,就变成了一个流。
- 应用场景:多种数据源的数据类型一致,数据处理规则也一致
scala代码:
package com.imooc.scala.streamimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}/*** 合并多个流,多个流的数据类型必须一致* 应用场景:多种数据源的数据类型一致,数据处理规则也一致**/
object StreamUnionScala {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._//第一份数据流val text1: DataStream[Int] = env.fromCollection(Array(1, 2, 3, 4, 5))//第二份数据流val text2: DataStream[Int] = env.fromCollection(Array(6, 7, 8, 9, 10))//合并流val unionStream: DataStream[Int] = text1.union(text2)//打印流中的数据unionStream.print().setParallelism(1)env.execute("StreamUnionScala")}
}
Java代码
package com.imooc.java.stream;import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Arrays;public class StreamUnionJava {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//第一份数据流DataStreamSource<Integer> text1 = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5));//第二份数据流DataStreamSource<Integer> text2 = env.fromCollection(Arrays.asList(6, 7, 8, 9, 10));//合并流DataStream<Integer> unionStream = text1.union(text2);//打印unionStream.print().setParallelism(1);env.execute("StreamUnionJava");}
}
2. connect:只能连接两个流,两个流的数据类型可以不同
两个流被connect之后,只是被放到了同一个流中,它们内部依然保持各自的数据和形式不发生任何变化,两个流相互独立。
connect方法会返回connectedStream,在connectedStream中需要使用CoMap、CoFlatMap这种函数,类似于map和flatmap。
Scala代码:
package com.imooc.scala.streamimport org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment}/*** 只能连接两个流,两个流的数据类型可以不同* 应用:可以将两种不同格式的数据统一成一种格式**/
object StreamConnectScala {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._val text1: DataStream[String] = env.fromElements("user:tom,age:18")val text2: DataStream[String] = env.fromElements("user:tom_age:18")//连接两个流val connectStream: ConnectedStreams[String, String] = text1.connect(text2)val text: DataStream[String] = connectStream.map(new CoMapFunction[String, String, String] {override def map1(in1: String): String = {in1.replace(",", "-")}override def map2(in2: String): String = {in2.replace("_", "-")}})//打印text.print().setParallelism(1)//env.execute("StreamConnectScala")}
}
Java代码:
package com.imooc.java.stream;import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;import java.awt.event.TextEvent;/*** 只能连接两个流,两个流的数据类型可以不同* 应用:可以将两种不同格式的数据统一成一种格式**/
public class StreamConnectJava {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//第1份数据流DataStreamSource<String> text1 = env.fromElements("user:tom,age:18");//第2份数据流DataStreamSource<String> text2 = env.fromElements("user:tom_age:18");//合并两个流ConnectedStreams<String, String> connectStream = text1.connect(text2);SingleOutputStreamOperator<Object> text = connectStream.map(new CoMapFunction<String, String, Object>() {@Overridepublic Object map1(String s) throws Exception {return s.replace(",", "-");}@Overridepublic Object map2(String s) throws Exception {return s.replace("_", "-");}});text.print().setParallelism(1);env.execute("StreamConnectJava");}
}
3.split:根据规则把一个数据流切分为多个流
注意:
(1)split只能分一次流,切分出来的流不能继续分流
(2)split需要和select配合使用,选择切分后的流
Scala代码:
package com.imooc.scala.streamimport java.{lang, util}import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment/*** 根据规则把一个数据流切分为多个流* 注意:split只能分一次流,切分出来的流不能继续分流* split需要和select配合使用,选择切分后的流* 应用场景:将一份数据流切分为多份,便于针对每一份数据使用不同的处理逻辑*/
object StreamSplitScala {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._val text = env.fromCollection(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))//按照数据的奇偶性对数据进行分流val splitStream = text.split(new OutputSelector[Int] {override def select(value: Int): lang.Iterable[String] = {val list = new util.ArrayList[String]()if (value % 2 == 0) {list.add("even") // 偶数} else {list.add("odd") //奇数}list}})//选择流val evenStream = splitStream.select("even")evenStream.print().setParallelism(1)val oddStream = splitStream.select("odd")oddStream.print().setParallelism(1)env.execute("StreamSplitScala")}
}
Java代码:
package com.imooc.java.stream;import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.ArrayList;
import java.util.Arrays;public class StreamSplitJava {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<Integer> text = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));SplitStream<Integer> splitStream = text.split(new OutputSelector<Integer>() {@Overridepublic Iterable<String> select(Integer value) {ArrayList<String> list = new ArrayList<>();if (value % 2 == 0) {list.add("even");} else {list.add("odd");}return list;}});DataStream<Integer> evenStream = splitStream.select("even");evenStream.print().setParallelism(1);DataStream<Integer> oddStream = splitStream.select("odd");oddStream.print().setParallelism(1);env.execute("StreamSplitJava");}
}
目前split切分的流无法进行二次切分,并且split方法已经标记为过时了,官方不推荐使用,现在官方推荐使用side output的方式实现。
下面我来看一下使用side output如何实现流的多次切分
Scala代码:
package com.imooc.scala.streamimport java.{lang, util}import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector/*** 根据规则把一个数据流切分为多个流* 注意:split只能分一次流,切分出来的流不能继续分流* split需要和select配合使用,选择切分后的流* 应用场景:将一份数据流切分为多份,便于针对每一份数据使用不同的处理逻辑*/
object StreamSplit2Scala {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentimport org.apache.flink.api.scala._val text = env.fromCollection(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))//按照数据的奇偶性对数据进行分流//首先定义两个sideoutput来准备保存切分出来的数据val outputTag1 = new OutputTag[Int]("even")val outputTag2 = new OutputTag[Int]("odd")//注意: process属于Flink中的低级apival outputStream = text.process(new ProcessFunction[Int, Int] {override def processElement(value: Int, context: ProcessFunction[Int, Int]#Context, collector: Collector[Int]): Unit = {if (value % 2 == 0) {context.output(outputTag1, value)} else {context.output(outputTag2, value)}}})val evenStream = outputStream.getSideOutput(outputTag1)val oddStream = outputStream.getSideOutput(outputTag2)
// evenStream.print().setParallelism(1)//对evenStream流进行二次切分val outputTag11 = new OutputTag[Int]("low") //保留小于等于5的数字val outputTag12 = new OutputTag[Int]("high") //保留大于5的数字val subOutputStream = evenStream.process(new ProcessFunction[Int, Int] {override def processElement(value: Int, context: ProcessFunction[Int, Int]#Context, collector: Collector[Int]): Unit = {if (value <= 5) {context.output(outputTag11, value)} else {context.output(outputTag12, value)}}})//获取小于等于5的数据流val lowStream = subOutputStream.getSideOutput(outputTag11)//获取大于5的数据流val highStream = subOutputStream.getSideOutput(outputTag12)lowStream.print().setParallelism(1)env.execute("StreamSplitScala")}
}
4. union和connect的区别
区别
(1)union可以连接多个流,最后汇总成一个流,流里面的数据使用相同的计算规则
(2)connect值可以连接2个流,最后汇总成一个流,但是流里面的两份数据相互还是独立的,每一份数据使用一个计算规则
5. split与side output区别
区别:
(1)如果是只需要切分一次的话使用split或者side output都可以
(2)如果想要切分多次,就不能使用split了,需要使用side output
6. 分区相关的算子
(三)、DataStream API之DataSink
DataSink是 输出组件,负责把计算好的数据输出到其它存储介质中
Flink支持把流数据输出到文件中,不过在实际工作中这种场景不多,因为流数据处理之后一般会存储到一些消息队列里面,或者数据库里面,很少会保存到文件中的。
还有就是print,直接打印,这个其实我们已经用了很多次了,这种用法主要是在测试的时候使用的,方便查看输出的结果信息
1. Flink提供了一批Connectors,可以实现输出到第三方目的地
2. 针对sink的这些connector,我们在实际工作中最常用的是kafka、redis,针对Flink提供的常用sink组件,可以提供这些容错性保证
3. 需求:接收Socket传输过来的数据,把数据保存到Redis的list队列中。
Scala代码:
package com.imooc.scala.sinkimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}object StreamRedisSinkScala {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//连接socket获取输入数据val text: DataStream[String] = env.socketTextStream("bigdata01", 9001)import org.apache.flink.api.scala._val listData: DataStream[(String, String)] = text.map(word => ("l_words_scana", word))//指定redisSinkval conf: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("bigdata01").setPort(6379).build()val redisSink = new RedisSink[Tuple2[String, String]](conf, new MyReidsMapper)listData.addSink(redisSink)env.execute("StreamRedisSinkScala")}}class MyReidsMapper extends RedisMapper[Tuple2[String, String]]{override def getCommandDescription: RedisCommandDescription = {new RedisCommandDescription(RedisCommand.LPUSH)}override def getKeyFromData(t: (String, String)): String = {t._1}override def getValueFromData(t: (String, String)): String = {t._2}
}
注意:执行代码之前,需要先开启socket和redis服务