备案控制台
探索云世界
开发者社区 数据库 文章 正文

Flink初试——对接Kafka

2023-07-07 199
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
云数据库 RDS MySQL Serverless，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
简介： Flink初试——对接Kafka

本篇文章我们用 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如下：

<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


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-0
20/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 closed
Processed a total of 1200 messages


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

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

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的方式运行作业

[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: root
20/11/30 01:40:16 INFO cli.CliFrontend:  Current Hadoop/Kerberos user: hdfs
20/11/30 01:40:16 INFO cli.CliFrontend:  JVM: Java HotSpot(TM) 64-Bit Server VM - Oracle Corporation - 1.8/25.171-b11
20/11/30 01:40:16 INFO cli.CliFrontend:  Maximum heap size: 3531 MiBytes
20/11/30 01:40:16 INFO cli.CliFrontend:  JAVA_HOME: /usr/java/latest
20/11/30 01:40:16 INFO cli.CliFrontend:  Hadoop version: 2.7.5
20/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.log
20/11/30 01:40:16 INFO cli.CliFrontend:     -Dlog4j.configuration=file:/etc/flink/conf/log4j-cli.properties
20/11/30 01:40:16 INFO cli.CliFrontend:     -Dlogback.configurationFile=file:/etc/flink/conf/logback.xml
20/11/30 01:40:16 INFO cli.CliFrontend:  Program Arguments:
20/11/30 01:40:16 INFO cli.CliFrontend:     run
20/11/30 01:40:16 INFO cli.CliFrontend:     -m
20/11/30 01:40:16 INFO cli.CliFrontend:     yarn-cluster
20/11/30 01:40:16 INFO cli.CliFrontend:     -c
20/11/30 01:40:16 INFO cli.CliFrontend:     com.iiot.alarm.InSufficientOilAlarms
20/11/30 01:40:16 INFO cli.CliFrontend:     /data0/flinkdemo/stream-1.0-SNAPSHOT-jar-with-dependencies.jar
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:java.io.tmpdir=/tmp
20/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=Linux
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.arch=amd64
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.version=3.10.0-327.el7.x86_64
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:user.name=root
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:user.home=/root
20/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/bin
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.free=134MB
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.max=3531MB
20/11/30 01:40:51 INFO zookeeper.ZooKeeper: Client environment:os.memory.total=359MB
20/11/30 01:40:51 INFO utils.Compatibility: Using emulated InjectSessionExpiration
20/11/30 01:40:51 INFO imps.CuratorFrameworkImpl: Starting
20/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@1460c81d
20/11/30 01:40:51 INFO common.X509Util: Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation
20/11/30 01:40:51 INFO zookeeper.ClientCnxnSocket: jute.maxbuffer value is 4194304 Bytes
20/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:2181
20/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:2181
20/11/30 01:40:51 ERROR curator.ConnectionState: Authentication failed
20/11/30 01:40:51 INFO imps.CuratorFrameworkImpl: Default schema
20/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 = 60000
20/11/30 01:40:51 INFO state.ConnectionStateManager: State change: CONNECTED
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: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已经插入数据了。


文章标签：
云消息队列 Kafka 版
微服务引擎
云数据库 RDS MySQL 版
实时计算 Flink版
消息中间件
BI
流计算
关系型数据库
资源调度
MySQL
Kafka
关键词：
云消息队列 Kafka 版flink
实时计算 Flink版对接
云消息队列 Kafka 版对接
预测性维护
目录
相关文章
你鞋带开了~
|
2月前
|
消息中间件 缓存 关系型数据库
Flink CDC产品常见问题之upsert-kafka增加参数报错如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
你鞋带开了~
203 2
cuicuicuic
|
3月前
|
消息中间件 关系型数据库 MySQL
Flink问题子实现Kafka到Mysql如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
cuicuicuic
379 2
sunrr
|
3月前
|
消息中间件 Kafka Apache
Flink 提供了与 Kafka 集成的官方 Connector，使得 Flink 能够消费 Kafka 数据
【2月更文挑战第6天】Flink 提供了与 Kafka 集成的官方 Connector，使得 Flink 能够消费 Kafka 数据
sunrr
74 2
aliyun4381607004
|
24天前
|
消息中间件 SQL Java
阿里云Flink-自定义kafka format实践及踩坑记录（以protobuf为例）
阿里云Flink-自定义kafka format实践及踩坑记录（以protobuf为例）
aliyun4381607004
774 3
三分钟热度的鱼
|
2月前
|
消息中间件 关系型数据库 MySQL
Flink CDC产品常见问题之用upsert的方式写入kafka失败如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
三分钟热度的鱼
53 0
三分钟热度的鱼
|
2月前
|
消息中间件 关系型数据库 Kafka
Flink CDC产品常见问题之Flink CDC里从kafka消费的时候顺序混乱如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
三分钟热度的鱼
51 0
三分钟热度的鱼
|
3月前
|
分布式计算 资源调度 Hadoop
Flink报错问题之Sql往kafka表写聚合数据报错如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
三分钟热度的鱼
44 3
三分钟热度的鱼
|
3月前
|
SQL 消息中间件 关系型数据库
Flink CDC数据同步问题之向kafka同步数据报错如何解决
Flink CDC数据同步是指利用Flink CDC实现不同数据源之间的实时数据同步任务；本合集旨在提供Flink CDC数据同步的操作指南、性能优化建议和常见问题处理，助力用户高效实施数据同步。
三分钟热度的鱼
37 0
sunrr
|
3月前
|
消息中间件 Kafka Apache
Apache Flink 是一个开源的分布式流处理框架
Apache Flink 是一个开源的分布式流处理框架
sunrr
487 5
灵杰开发者
|
2月前
|
SQL Java API
官宣｜Apache Flink 1.19 发布公告
Apache Flink PMC（项目管理委员）很高兴地宣布发布 Apache Flink 1.19.0。
灵杰开发者
1371 2
官宣｜Apache Flink 1.19 发布公告

热门文章

最新文章

  • 1
    云原生最佳实践系列 5：基于函数计算 FC 实现阿里云 Kafka 消息内容控制 MongoDB DML 操作
  • 2
    【Kafka】分区副本中的 Leader 如果宕机但 ISR 却为空该如何处理
  • 3
    【Kafka】Kafka 消息的消费模式
  • 4
    kafka基本知识概述
  • 5
    【Kafka】Kafka 性能高的原因分析
  • 6
    【Kafka】Kafka 分区
  • 7
    Kafka消费者：监听模式VS主动拉取，哪种更适合你？
  • 8
    Kafka重要配置参数全面解读(重要)
  • 9
    【Kafka】Kafka 数据一致性原理
  • 10
    深度解析Kafka中的消息奥秘
  • 1
    在Kubernetes上运行Flink应用程序时
    38
  • 2
    Flink CDC产品常见问题之大文件整库同步怎么解决
    74
  • 3
    Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
    41
  • 4
    Flink CDC产品常见问题之look up hint 没有生效如何解决
    808
  • 5
    Flink CDC产品常见问题之写hudi的时候报错如何解决
    406
  • 6
    Flink CDC产品常见问题之tidb cdc 数据量大了就疯狂报空指针如何解决
    440
  • 7
    Flink CDC产品常见问题之pg cdc程序已经停了但是执行删不掉如何解决
    234
  • 8
    Flink CDC产品常见问题之没有报错但是一直监听不到数据如何解决
    48
  • 9
    Flink CDC产品常见问题之flink集群jps命令报错如何解决
    32
  • 10
    Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
    66

    • 相关课程

    更多
  • Apache Flink 入门到实战 - Flink开源社区出品
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 分布式消息系统 Kafka 快速入门
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

    更多
  • Flink CDC Meetup PPT - 龚中强
  • Flink CDC Meetup PPT - 王赫
  • Flink CDC Meetup PPT - 覃立辉

    • 相关实验场景

    更多
  • 基于Hologres轻松玩转一站式实时仓库
  • 数据湖构建DLF快速入门
  • 通过FastMR自动拉起大数据集群并运行TPCDS任务
  • 使用阿里云Elasticsearch快速搭建可观测系统
  • 消息队列+Serverless+Tablestore：实现高弹性的电商订单系统
  • EMR数据湖开发治理之用户画像分析
    • 下一篇
    2024年阿里云免费云服务器及学生云服务器申请教程参考