Excel实战解析之项目进度图(excel做进度图)
791
2022-05-30
欢迎访问我的GitHub
这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos
版本和环境准备
本次实战的环境和版本如下:
JDK:1.8.0_211
Flink:1.9.2
Maven:3.6.0
操作系统:macOS Catalina 10.15.3 (MacBook Pro 13-inch, 2018)
IDEA:2018.3.5 (Ultimate Edition)
Kafka:2.4.0
Zookeeper:3.5.5
请确保上述环境和服务已经就绪;
源码下载
如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blog_demos):
这个git项目中有多个文件夹,本章的应用在
flinksinkdemo
文件夹下,如下图红框所示:
准备完毕,开始开发;
准备工作
正式编码前,先去官网查看相关资料了解基本情况:
地址:https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html
我这里用的kafka是2.4.0版本,在官方文档查找对应的库和类,如下图红框所示:
kafka准备
创建名为test006的topic,有四个分区,参考命令:
./kafka-topics.sh \ --create \ --bootstrap-server 127.0.0.1:9092 \ --replication-factor 1 \ --partitions 4 \ --topic test006
在控制台消费test006的消息,参考命令:
./kafka-console-consumer.sh \ --bootstrap-server 127.0.0.1:9092 \ --topic test006
此时如果该topic有消息进来,就会在控制台输出;
接下来开始编码;
创建工程
用maven命令创建flink工程:
mvn \ archetype:generate \ -DarchetypeGroupId=org.apache.flink \ -DarchetypeArtifactId=flink-quickstart-java \ -DarchetypeVersion=1.9.2
根据提示,groupid输入
com.bolingcavalry
,artifactid输入
flinksinkdemo
,即可创建一个maven工程;
在pom.xml中增加kafka依赖库:
工程创建完成,开始编写flink任务的代码;
发送字符串消息的sink
先尝试发送字符串类型的消息:
创建KafkaSerializationSchema接口的实现类,后面这个类要作为创建sink对象的参数使用:
package com.bolingcavalry.addsink; import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema; import org.apache.kafka.clients.producer.ProducerRecord; import java.nio.charset.StandardCharsets; public class ProducerStringSerializationSchema implements KafkaSerializationSchema
创建任务类KafkaStrSink,请注意FlinkKafkaProducer对象的参数,FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次:
package com.bolingcavalry.addsink; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import java.util.ArrayList; import java.util.List; import java.util.Properties; public class KafkaStrSink { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //并行度为1 env.setParallelism(1); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "192.168.50.43:9092"); String topic = "test006"; FlinkKafkaProducer
使用mvn命令编译构建,在target目录得到文件
flinksinkdemo-1.0-SNAPSHOT.jar
;
在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar,并制定执行类,如下图:
提交成功后,如果flink有四个可用slot,任务会立即执行,会在消费kafak消息的终端收到消息,如下图:
任务执行情况如下图:
发送对象消息的sink
再来尝试如何发送对象类型的消息,这里的对象选择常用的Tuple2对象:
创建KafkaSerializationSchema接口的实现类,该类后面要用作sink对象的入参,请注意代码中捕获异常的那段注释:
生产环境慎用printStackTrace()!!!
package com.bolingcavalry.addsink; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema; import org.apache.kafka.clients.producer.ProducerRecord; import javax.annotation.Nullable; public class ObjSerializationSchema implements KafkaSerializationSchema
创建flink任务类:
package com.bolingcavalry.addsink; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import java.util.ArrayList; import java.util.List; import java.util.Properties; public class KafkaObjSink { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //并行度为1 env.setParallelism(1); Properties properties = new Properties(); //kafka的broker地址 properties.setProperty("bootstrap.servers", "192.168.50.43:9092"); String topic = "test006"; FlinkKafkaProducer
像前一个任务那样编译构建,把jar提交到flink,并指定执行类是
com.bolingcavalry.addsink.KafkaObjSink
;
消费kafka消息的控制台输出如下:
在web页面可见执行情况如下:
至此,flink将计算结果作为kafka消息发送出去的实战就完成了,希望能给您提供参考,接下来的章节,我们会继续体验官方提供的sink能力;
欢迎关注华为云博客:程序员欣宸
学习路上,你不孤单,欣宸原创一路相伴…
Flink Kafka
版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。