前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Flink初试——对接Kafka

Flink初试——对接Kafka

作者头像
soundhearer
发布2021-01-06 15:14:11
2K0
发布2021-01-06 15:14:11
举报
文章被收录于专栏:数据湖

本篇文章我们用 Flink Kafka Connector对接Kafka,实现一个简单的报警业务。我们暂时不去谈论理论,先上手实现这个简单的需求。

flink-connector-kafka是 flink 内置的Kafka连接器,包含了从topic读取数据的Flink Kafka Consumer 和 向topic写入数据的flink kafka producer,除了基本功能外还提供了基于 checkpoint 机制提供了完美的容错能力。

本文基于flink 1.10.1 和 flink-connector-kafka-0.10_2.11版本,pom如下:

代码语言:javascript
复制
<dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-connector-kafka-0.10_2.11</artifactId>            <version>1.10.0</versio></dependency>

以企业常见的预警业务为例,本文要实现的业务逻辑很简单,当设备上报的油桶余量不足10%时,便生成一个报警,这里我们将报警写入MySQL,以供web业务端展示报警报表。

首先我们用网络数据调试器向网关模拟发送数据,网关会将数据解析后写入kafka

代码语言:javascript
复制
kafka-console-consumer --bootstrap-server cdh1.macro.com:9092,cdh2.macro.com:9092,cdh3.macro.com:9092 --from-beginning --topic fill
{"addTime":1593147840000,"currentAmount":0.3,"devId":"XT365-000170","devStatus":"1","ifOffline":"1","ip":"127.0.0.1","leftTankAmount":5,"realTotalAmount":2377.39,"registerTime":1606658457000,"settingAmount":0.3,"tankCapacity":1000,"totalAmount":2017.9315}{"addTime":1593147840000,"currentAmount":0.3,"devId":"XT365-000170","devStatus":"1","ifOffline":"1","ip":"127.0.0.1","leftTankAmount":5,"realTotalAmount":2377.69,"registerTime":1606658458000,"settingAmount":0.3,"tankCapacity":1000,"totalAmount":2017.9315}^C20/11/29 23:26:55 INFO internals.ConsumerCoordinator: [Consumer clientId=consumer-console-consumer-82199-1, groupId=console-consumer-82199] Revoke previously assigned partitions fill-020/11/29 23:26:55 INFO internals.AbstractCoordinator: [Consumer clientId=consumer-console-consumer-82199-1, groupId=console-consumer-82199] Member consumer-console-consumer-82199-1-aa5fc2e6-1f06-4714-9d89-fe080a9400e2 sending LeaveGroup request to coordinator cdh2.macro.com:9092 (id: 2147483598 rack: null) due to the consumer is being closedProcessed a total of 1200 messages

可以看到我们已经向kafka生产了1200条数据了

接下来我们写一段代码来消费kafka数据,并将报警结果写入MySQL

代码语言:javascript
复制
import com.alibaba.fastjson.JSONObject;import com.iiot.bean.InSufficient;import com.iiot.commCommon.Fill;import com.iiot.jdbc.MySQLSinks;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.streaming.api.datastream.DataStream;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.windowing.windows.TimeWindow;import org.apache.flink.streaming.api.windowing.time.Time;import java.util.List;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;import org.apache.flink.util.Collector;import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import java.util.Properties;
public class InSufficientOilAlarms {    public static void main(String[] args) throws Exception{        //构建流执行环境        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //kafka        Properties prop = new Properties();        prop.put("bootstrap.servers", "cdh1.macro.com:9092,cdh2.macro.com:9092,cdh3.macro.com:9092");//        prop.put("zookeeper.connect", "localhost:2181");        prop.put("group.id", "fill6");        prop.put("key.serializer", "org.apache.kafka.common.serialization.StringDeserializer");        prop.put("value.serializer", "org.apache.kafka.common.serialization.StringDeserializer");        prop.put("auto.offset.reset", "earliest");
        DataStreamSource<String> stream = env                .addSource(new FlinkKafkaConsumer010<String>(                        "fill",                        new SimpleStringSchema(), prop)).                //单线程打印,控制台不乱序,不影响结果                setParallelism(1);
        //从kafka里读取数据,转换成Person对象        DataStream<Fill> dataStream = stream.map(value ->                JSONObject.parseObject(value, Fill.class)        );
        SingleOutputStreamOperator<InSufficient> result = dataStream.map(new MapFunction<Fill, InSufficient>() {                           @Override                           public InSufficient map(Fill fill) throws Exception {                               InSufficient inSufficient = new InSufficient();                               Float leftTankAmount = fill.getLeftTankAmount();                               Float tankCapacity = fill.getTankCapacity();                               String devCode = fill.getDevId();                               long timeBegin = fill.getAddTime().getTime();                               System.out.println("devCode:-------------------------------------------------" + devCode);                               String alarmType = "";                               if ((leftTankAmount / tankCapacity) < 0.1 ) {                                   alarmType = "inSufficientOil";                                   inSufficient.setDev_code(devCode);                                   inSufficient.setCreateTime(System.currentTimeMillis());                                   inSufficient.setTimeBegin(timeBegin);                                   inSufficient.setAlarmType(alarmType);                                   inSufficient.setRemainAmount(leftTankAmount);                               }                               return inSufficient;                           }                       }
        );

        //收集5秒钟的总数        result.timeWindowAll(Time.seconds(5L)).                apply(new AllWindowFunction<InSufficient, List<InSufficient>, TimeWindow>() {
                    @Override                    public void apply(TimeWindow timeWindow, Iterable<InSufficient> iterable, Collector<List<InSufficient>> out) throws Exception {                        List<InSufficient> inSufficients = Lists.newArrayList(iterable);
                        if(inSufficients.size() > 0) {                            System.out.println("5秒的总共收到的条数:" + inSufficients.size());                            out.collect(inSufficients);                        }
                    }                })                //sink 到数据库                .addSink(new MySQLSinks());        //打印到控制台        //.print();
        env.execute("kafka 消费任务开始");    }}

将项目打包,传到集群中,用Flink on YARN的方式运行作业

代码语言:javascript
复制
/*
* 提示:该行代码过长,系统自动注释不进行高亮。一键复制会移除系统注释 
* [root@cdh3 bin]# flink run -m yarn-cluster -c com.iiot.alarm.InSufficientOilAlarms /data0/flinkdemo/stream-1.0-SNAPSHOT-jar-with-dependencies.jar 20/11/30 01:40:15 INFO cli.CliFrontend: --------------------------------------------------------------------------------20/11/30 01:40:15 INFO cli.CliFrontend:  Starting Command Line Client (Version: 1.10.0-csa1.2.0.0, Rev:04dddd1, Date:29.05.2020 @ 14:54:45 UTC)20/11/30 01:40:15 INFO cli.CliFrontend:  OS current user: root20/11/30 01:40:16 INFO cli.CliFrontend:  Current Hadoop/Kerberos user: hdfs20/11/30 01:40:16 INFO cli.CliFrontend:  JVM: Java HotSpot(TM) 64-Bit Server VM - Oracle Corporation - 1.8/25.171-b1120/11/30 01:40:16 INFO cli.CliFrontend:  Maximum heap size: 3531 MiBytes20/11/30 01:40:16 INFO cli.CliFrontend:  JAVA_HOME: /usr/java/latest20/11/30 01:40:16 INFO cli.CliFrontend:  Hadoop version: 2.7.520/11/30 01:40:16 INFO cli.CliFrontend:  JVM Options:20/11/30 01:40:16 INFO cli.CliFrontend:     -Datlas.conf=/etc/atlas/conf/20/11/30 01:40:16 INFO cli.CliFrontend:     -Dlog.file=/var/log/flink/flink-root-client-cdh3.macro.com.log20/11/30 01:40:16 INFO cli.CliFrontend:     -Dlog4j.configuration=file:/etc/flink/conf/log4j-cli.properties20/11/30 01:40:16 INFO cli.CliFrontend:     -Dlogback.configurationFile=file:/etc/flink/conf/logback.xml20/11/30 01:40:16 INFO cli.CliFrontend:  Program Arguments:20/11/30 01:40:16 INFO cli.CliFrontend:     run20/11/30 01:40:16 INFO cli.CliFrontend:     -m20/11/30 01:40:16 INFO cli.CliFrontend:     yarn-cluster20/11/30 01:40:16 INFO cli.CliFrontend:     -c20/11/30 01:40:16 INFO cli.CliFrontend:     com.iiot.alarm.InSufficientOilAlarms20/11/30 01:40:16 INFO cli.CliFrontend:     /data0/flinkdemo/stream-1.0-SNAPSHOT-jar-with-dependencies.jar20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:java.io.tmpdir=/tmp20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:java.compiler=<NA>20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.name=Linux20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.arch=amd6420/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.version=3.10.0-327.el7.x86_6420/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:user.name=root20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:user.home=/root20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:user.dir=/opt/cloudera/parcels/FLINK-1.10.0-csa1.2.0.0-cdh7.1.1.0-565-3454809/bin20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.free=134MB20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.max=3531MB20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.total=359MB20/11/30 01:40:51 INFO utils.Compatibility: Using emulated InjectSessionExpiration20/11/30 01:40:51 INFO imps.CuratorFrameworkImpl: Starting20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Initiating client connection, connectString=cdh1.macro.com:2181,cdh2.macro.com:2181,cdh3.macro.com:2181 sessionTimeout=60000 watcher=org.apache.flink.shaded.curator.org.apache.curator.ConnectionState@1460c81d20/11/30 01:40:51 INFO common.X509Util: Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation20/11/30 01:40:51 INFO zookeeper.ClientCnxnSocket: jute.maxbuffer value is 4194304 Bytes20/11/30 01:40:51 INFO zookeeper.ClientCnxn: zookeeper.request.timeout value is 0. feature enabled=20/11/30 01:40:51 WARN zookeeper.ClientCnxn: SASL configuration failed: javax.security.auth.login.LoginException: No JAAS configuration section named 'Client' was found in specified JAAS configuration file: '/tmp/jaas-8202592158525653501.conf'. Will continue connection to Zookeeper server without SASL authentication, if Zookeeper server allows it.20/11/30 01:40:51 INFO zookeeper.ClientCnxn: Opening socket connection to server cdh1.macro.com/192.168.0.171:218120/11/30 01:40:51 INFO zookeeper.ClientCnxn: Socket connection established, initiating session, client: /192.168.0.208:38183, server: cdh1.macro.com/192.168.0.171:218120/11/30 01:40:51 ERROR curator.ConnectionState: Authentication failed20/11/30 01:40:51 INFO imps.CuratorFrameworkImpl: Default schema20/11/30 01:40:51 INFO zookeeper.ClientCnxn: Session establishment complete on server cdh1.macro.com/192.168.0.171:2181, sessionid = 0x3008be9995512b4, negotiated timeout = 6000020/11/30 01:40:51 INFO state.ConnectionStateManager: State change: CONNECTED20/11/30 01:40:51 INFO imps.EnsembleTracker: New config event received: {server.1=cdh2.macro.com:3181:4181:participant, version=0, server.3=cdh1.macro.com:3181:4181:participant, server.2=cdh3.macro.com:3181:4181:participant}20/11/30 01:40:51 ERROR imps.EnsembleTracker: Invalid config event received: {server.1=cdh2.macro.com:3181:4181:participant, version=0, server.3=cdh1.macro.com:3181:4181:participant, server.2=cdh3.macro.com:3181:4181:participant}20/11/30 01:40:51 INFO imps.EnsembleTracker: New config event received: {server.1=cdh2.macro.com:3181:4181:participant, version=0, server.3=cdh1.macro.com:3181:4181:participant, server.2=cdh3.macro.com:3181:4181:participant}20/11/30 01:40:51 ERROR imps.EnsembleTracker: Invalid config event received: {server.1=cdh2.macro.com:3181:4181:participant, version=0, server.3=cdh1.macro.com:3181:4181:participant, server.2=cdh3.macro.com:3181:4181:participant}20/11/30 01:40:52 INFO leaderretrieval.ZooKeeperLeaderRetrievalService: Starting ZooKeeperLeaderRetrievalService /leader/rest_server_lock.
*/

可以在YARN作业中看到Flink的做作业一直在运行。

flink dashboard也可以看到作业一直在运行:

进入YARN reourcemanager里面查看作业运行日志:

可以看到MySQL已经插入数据了。

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-12-21,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 数据湖 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
数据库
云数据库为企业提供了完善的关系型数据库、非关系型数据库、分析型数据库和数据库生态工具。您可以通过产品选择和组合搭建,轻松实现高可靠、高可用性、高性能等数据库需求。云数据库服务也可大幅减少您的运维工作量,更专注于业务发展,让企业一站式享受数据上云及分布式架构的技术红利!
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档