<dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-scala_2.11</artifactId> <version>1.7.0</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.11</artifactId> <version>1.7.0</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka-0.11_2.12</artifactId> <version>1.7.0</version> </dependency> </dependencies> <build> <plugins> <!-- 该插件用于将Scala代码编译成class文件 --> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.4.6</version> <executions> <execution> <!-- 声明绑定到maven的compile阶段 --> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-assembly-plugin</artifactId> <version>3.0.0</version> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>
回到顶部
二、Source
1,基本读取方式
//文件中读取 val fileDs = env.readTextFile("in/tbStock.txt") //端口读取 val socketDs = env.socketTextStream("localhost",777) //集合中获取 val collectDs = env.fromCollection(List("aaa","bbb","ccc","aaa"))
2,kafka source
//kafka配置文件 val properties = new Properties() properties.setProperty("bootstrap.servers", "hadoop102: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") //接收kafka的topic-demo这个topic发来的数据 val kafkaDataStream: DataStream[String] = env.addSource(new FlinkKafkaConsumer011[String]("topic-demo", new SimpleStringSchema(), properties))
3,Flink Kafa如何实现exactly-once
可参考: https://www.aboutyun.com/forum.php?mod=viewthread&tid=27395
Flink通过checkpoint来保存数据是否处理完成的状态
由JobManager协调各个TaskManager进行checkpoint存储,checkpoint保存在 StateBackend中,默认StateBackend是内存级的,也可以改为文件级的进行持久化保存。
执行过程实际上是一个两段式提交,每个算子执行完成,会进行“预提交”,直到执行完sink操作,会发起“确认提交”,如果执行失败,预提交会放弃掉。
如果宕机需要通过StateBackend进行恢复,只能恢复所有确认提交的操作。
4,自定义source
env.addSource(new MySource) //自定义source class MySource extends SourceFunction[(String,Double)] { //flag: 表示数据源是否还在正常运行 var running: Boolean = true override def cancel(): Unit = { running = false } override def run(ctx: SourceFunction.SourceContext[(String,Double)]): Unit = { //初始化一个随机数发生器 val rand = new Random() var curTemp = 1.to(10).map( i => ("item_" + i, 65 + rand.nextGaussian() * 20) ) while (running) { curTemp.foreach( t => ctx.collect(t) ) Thread.sleep(5000) //每5秒钟产生一组数据 } } }
回到顶部
三、Transform
1,基本转换算子
//map val streamMap = stream.map { x => x * 2 } //flatmap val streamFlatMap = stream.flatMap{ x => x.split(" ") } //filter val streamFilter = stream.filter{ x => x == 1 }
2,KeyBy与Reduce
keyBy(DataStream → KeyedStream):输入必须是Tuple类型,逻辑地将一个流拆分成不相交的分区,每个分区包含具有相同key的元素,在内部以hash的形式实现的。
reduce(KeyedStream → DataStream):一个分组数据流的聚合操作,合并当前的元素和上次聚合的结果,产生一个新的值,返回的流中包含每一次聚合的结果,而不是只返回最后一次聚合的最终结果。
val keyedStream: KeyedStream[(String, Int), Tuple] = startUplogDstream.map(startuplog=>(startuplog.ch,1)).keyBy(0) //reduce //sum keyedStream.reduce{ (ch1,ch2)=> (ch1._1,ch1._2+ch2._2) }.print()
3,Split和Select
split(DataStream → SplitStream):根据某些特征把一个DataStream拆分成两个或者多个DataStream。
select(SplitStream→DataStream):从一个SplitStream中获取一个或者多个DataStream。
//根据Item的id进行拆分 val splitStream:SplitStream[Item] = dStream.split { item => List(item.id) } //获取标记为item_1的数据集 splitStream.select("item_1").print()
4,Connect和CoMap
connect(DataStream,DataStream → ConnectedStreams):连接两个保持他们类型的数据流,两个数据流被Connect之后,只是被放在了一个同一个流中,内部依然保持各自的数据和形式不发生任何变化,两个流相互独立。
CoMap,CoFlatMap(ConnectedStreams → DataStream):作用于ConnectedStreams上,功能与map和flatMap一样,对ConnectedStreams中的每一个Stream分别进行map和flatMap处理。
val connStream: ConnectedStreams[StartUpLog, StartUpLog] = appStoreStream.connect(otherStream) val allStream: DataStream[String] = connStream.map( (log1: StartUpLog) => log1.ch, (log2: StartUpLog) => log2.ch )
4,Union
DataStream → DataStream:对两个或者两个以上的DataStream进行union操作,产生一个包含所有DataStream元素的新DataStream。注意:如果你将一个DataStream跟它自己做union操作,在新的DataStream中,你将看到每一个元素都出现两次。
val unionStream: DataStream[StartUpLog] = appStoreStream.union(otherStream) unionStream.print("union:::")
5,Connect与 Union 区别:
1)Union之前两个流的类型必须是一样,Connect可以不一样,在之后的coMap中再去调整成为一样的。
2)Connect只能操作两个流,Union可以操作多个
回到顶部
四、实现UDF函数
1,函数类(Function Classes)
Flink暴露了所有udf函数的接口(实现方式为接口或者抽象类)。例如:MapFunction, FilterFunction, ProcessFunction 等等。
val flinkTweets = tweets.filter(new FlinkFilter)
//自定义filter类 class FlinkFilter extends FilterFunction[String] { override def filter(value: String): Boolean = { value.contains("flink") } }
2,匿名函数(Lamda Functions)
val flinkTweets = tweets.filter(_.contains("flink"))
3,富含数(Rich Functions)
富函数是 DataStream API 提供的一个函数类的接口,所有 Flink 函数类都有其 Rich 版本。它与常规函数的不同在于,可以获取运行环境的上下文,并拥有一些生命周期方法,所以可以实现更复杂的功能。
open()方法是 rich function 的初始化方法,当一个算子例如map或者filter被调用之前open()会被调用。
close()方法是生命周期中的最后一个调用的方法,做一些清理工作。
getRuntimeContext()方法提供了函数的 RuntimeContext 的一些信息,例如函数执行的并行度,任务的名字,以及 state 状态。
回到顶部
五、Sink
Flink 没有类似于spark中foreach方法,让用户进行迭代的操作。虽有对外的输出操作都要利用Sink完成。最后通过类似如下方式完成整个任务最终输出操作。
1,kafka
dstream.addSink(new FlinkKafkaProducer011[String]("linux01:9092","test", new SimpleStringSchema()))
2,redis
<dependency> <groupId>org.apache.bahir</groupId> <artifactId>flink-connector-redis_2.11</artifactId> <version>1.0</version> </dependency>
val config = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort(6379).build() resultDStream.addSink(new RedisSink[Item](config,new MyRedisMapper)) //定义redisMapper class MyRedisMapper extends RedisMapper[Item] { override def getCommandDescription: RedisCommandDescription = { new RedisCommandDescription(RedisCommand.HSET,"item_test") //hkey } override def getKeyFromData(data: Item): String = data.id override def getValueFromData(data: Item): String = data.toString }
3,Elasticsearch
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-elasticsearch6_2.11</artifactId> <version>1.7.2</version> </dependency>
//定义es的host集合 val list = new util.ArrayList[HttpHost]() list.add(new HttpHost("linux01", 9200)) //定义esBuilder val esBuilder = new ElasticsearchSink.Builder[Item](list,new ElasticsearchSinkFunction[Item] { override def process(element: Item, ctx: RuntimeContext, indexer: RequestIndexer): Unit = { //定义es数据存储方式和存储值 val json = new util.HashMap[String, String]() json.put("data", element.toString) //定义存储索引 type 和数据源 val indexRequest = Requests.indexRequest().index("indexName").`type`("_doc").source(json) indexer.add(indexRequest) } }) resultDStream.addSink(esBuilder.build())
4,自定义sink(JDBC)
<dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.44</version> </dependency>
resultDStream.addSink(new MyJDBCSink) //自定义jdbcsink class MyJDBCSink extends RichSinkFunction[Sensor]{ var conn: Connection = _ var insertStmt: PreparedStatement = _ var updateStmt: PreparedStatement = _ //open 简历连接 override def open(parameters: Configuration): Unit = { conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456") insertStmt = conn.prepareStatement("INSERT INTO item_test (id, num) VALUES (?, ?)") updateStmt = conn.prepareStatement("UPDATE item_test SET num = ? WHERE id = ?") } //调用执行 override def invoke(value: Sensor, context: SinkFunction.Context[_]): Unit = { updateStmt.setDouble(1, value.temp) updateStmt.setString(2, value.id) updateStmt.execute() if (updateStmt.getUpdateCount == 0) { insertStmt.setString(1, value.id) insertStmt.setDouble(2, value.temp) insertStmt.execute() } } //关闭资源 override def close(): Unit = { insertStmt.close() updateStmt.close() conn.close() } }