Flink实战 —— 读取Kafka数据并与MySQL数据关联【附源码】

目录

一、功能需求说明

二、前期准备工作

2.1 需要导入mysql驱动

2.2 mysql建立表及插入数据

2.3 启动Kafka的topic

2.4 编写 Flink相关应用代码

三、 启动Flink 应用程序及向Kafka生产数据 

3.1 可以在idea本地启动 C01_QueryActivityName 

3.2 通过向Kafka-producer生产数据

一、功能需求说明

Flink 从 Kafka消息中间件中取数据，有活动id 无活动名称Name，Flink 通过与MySQL关联到活动名称返回

二、前期准备工作

2.1 需要导入mysql驱动

  1.         
  2.             mysql
  3.             mysql-connector-java
  4.             5.1.44
  5.

2.2 mysql建立表及插入数据

  1. DROP DATABASE IF EXISTS `flink_big_data`; -- 库名与项目名保持一致
  2. CREATE DATABASE IF NOT EXISTS `flink_big_data` DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci;
  3.  
  4. USE `flink_big_data`;
  5.  
  6. -- 活动列表
  7. DROP TABLE IF EXISTS `t_activities`;
  8. CREATE TABLE `t_activities` (
  9.   `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键id, 必备字段',
  10.   `gmt_create` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间, 必备字段',
  11.   `gmt_modified` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间, 必备字段',
  12.  
  13.   `a_id` VARCHAR(100) NOT NULL COMMENT '活动id',
  14.   `name` VARCHAR(100) NOT NULL COMMENT '活动名称',
  15.   `last_update` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间, 必备字段',
  16.   PRIMARY KEY (`id`)
  17. ) ENGINE=INNODB DEFAULT CHARSET=utf8;
  18.  
  19. -- 插入数据
  20. INSERT INTO `t_activities` (`a_id`, `name`) VALUES ('A1', '新人礼包');
  21. INSERT INTO `t_activities` (`a_id`, `name`) VALUES ('A2', '月末活动');
  22. INSERT INTO `t_activities` (`a_id`, `name`) VALUES ('A3', '周末促销');
  23. INSERT INTO `t_activities` (`a_id`, `name`) VALUES ('A4', '年度促销');

2.3 启动Kafka的topic

  1. # 创建topic
  2. kafka-topics --create --zookeeper node-01:2181,node-02:2181,node-03:2181 --replication-factor 2 --partitions 3 --topic activity10
  3.  
  4. # 创建生产者
  5. kafka-console-producer --broker-list node-01:9092,node-02:9092,node-03:9092 --topic activity10

2.4 编写 Flink相关应用代码

  1. public class ActivityBean {
  2.  
  3.     public String uid; // userId
  4.     public String aid; // activityId
  5.     public String activityName;
  6.     public String time;
  7.     public int eventType;
  8.     public double longitude;
  9.     public double latitude;
  10.     public String province;
  11.     public int count = 1;
  12.  
  13.     public ActivityBean() {
  14.     }
  15.  
  16.     public ActivityBean(String uid, String aid, String activityName, String time, int eventType, String province) {
  17.         this.uid = uid;
  18.         this.aid = aid;
  19.         this.activityName = activityName;
  20.         this.time = time;
  21.         this.eventType = eventType;
  22.         this.province = province;
  23.     }
  24.  
  25.     public ActivityBean(String uid, String aid, String activityName, String time, int eventType, double longitude, double latitude, String province) {
  26.         this.uid = uid;
  27.         this.aid = aid;
  28.         this.activityName = activityName;
  29.         this.time = time;
  30.         this.eventType = eventType;
  31.         this.longitude = longitude;
  32.         this.latitude = latitude;
  33.         this.province = province;
  34.     }
  35.  
  36.     @Override
  37.     public String toString() {
  38.         return "ActivityBean{" +
  39.                 "uid='" + uid + '\'' +
  40.                 ", aid='" + aid + '\'' +
  41.                 ", activityName='" + activityName + '\'' +
  42.                 ", time='" + time + '\'' +
  43.                 ", eventType=" + eventType +
  44.                 ", longitude=" + longitude +
  45.                 ", latitude=" + latitude +
  46.                 ", province='" + province + '\'' +
  47.                 ", count=" + count +
  48.                 '}';
  49.     }
  50.  
  51.     public static ActivityBean of(String uid, String aid, String activityName, String time, int eventType, String province) {
  52.         return new ActivityBean(uid, aid, activityName, time, eventType, province);
  53.     }
  54.  
  55.     public static ActivityBean of(String uid, String aid, String activityName, String time, int eventType, double longitude, double latitude, String province) {
  56.         return new ActivityBean(uid, aid, activityName, time, eventType, longitude, latitude, province);
  57.     }
  58. }
  1. public class C01_DataToActivityBeanFunction extends RichMapFunction {
  2.  
  3.     private Connection connection = null;
  4.  
  5.     @Override
  6.     public void open(Configuration parameters) throws Exception {
  7.         super.open(parameters);
  8.  
  9.         // 创建MySQL连接
  10.         // 这里不应该对异常进行捕获，让Flink自行处理，比如重启之类的
  11.         // 如果捕获异常了，则Flink无法捕获到该异常
  12.         String url = "jdbc:mysql://localhost:3306/flink_big_data?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&failOverReadOnly=false";
  13.         String user = "root";
  14.         String password = "1234";
  15.         connection = DriverManager.getConnection(url, user, password);
  16.     }
  17.  
  18.     @Override
  19.     public ActivityBean map(String line) throws Exception {
  20.         String[] fields = line.split(",");
  21.  
  22.         String uid = fields[0];
  23.         String aid = fields[1];
  24.  
  25.         // 根据aid作为查询条件查询出name
  26.         // 最好使用简单的关联查询，MySQL也可以进行关联查询
  27.         PreparedStatement preparedStatement = connection.prepareStatement("SELECT name FROM t_activities WHERE a_id = ?");
  28.         preparedStatement.setString(1, aid);
  29.         ResultSet resultSet = preparedStatement.executeQuery();
  30.         String name = null;
  31.         while (resultSet.next()) {
  32.             name = resultSet.getString(1);
  33.         }
  34.  
  35.         String time = fields[2];
  36.         int eventType = Integer.parseInt(fields[3]);
  37.         String province = fields[4];
  38.  
  39.         return ActivityBean.of(uid, aid, name, time, eventType, province);
  40.     }
  41.  
  42.     @Override
  43.     public void close() throws Exception {
  44.         super.close();
  45.         if (connection != null) {
  46.             connection.close();
  47.         }
  48.     }
  49. }
  1. public class C01_QueryActivityName {
  2.     public static void main(String[] args) throws Exception {
  3.         // topic:activity10 分区3，副本2
  4.         // # 创建topic
  5.         // kafka-topics --create --zookeeper node-01:2181,node-02:2181,node-03:2181 --replication-factor 2 --partitions 3 --topic activity10
  6.  
  7.         // # 创建生产者
  8.         // kafka-console-producer --broker-list node-01:9092,node-02:9092,node-03:9092 --topic activity10
  9.  
  10.         // 输入参数：activity10 group_id_flink node-01:9092,node-02:9092,node-03:9092
  11.         DataStream lines = FlinkUtilsV1.createKafkaStream(args, new SimpleStringSchema());
  12.  
  13.         SingleOutputStreamOperator beans = lines.map(new C01_DataToActivityBeanFunction());
  14.  
  15.         beans.print();
  16.  
  17.         FlinkUtilsV1.getEnv().execute("C01_QueryActivityName");
  18.  
  19.     }
  20. }

三、 启动Flink 应用程序及向Kafka生产数据 

3.1 可以在idea本地启动 C01_QueryActivityName 

3.2 通过向Kafka-producer生产数据

  1. u001,A1,2019-09-02 10:10:11,1,北京市
  2. u001,A2,2019-09-02 10:10:11,1,北京市
  3. u001,A3,2019-09-02 10:10:11,1,北京市
  4. u001,A4,2019-09-02 10:10:11,1,北京市
  5. u002,A1,2019-09-02 10:11:11,1,辽宁省
  6. u001,A1,2019-09-02 10:11:11,2,北京市
  7. u001,A1,2019-09-02 10:11:30,3,北京市
  8. u002,A1,2019-09-02 10:12:11,2,辽宁省
  9. u001,A1,2019-09-02 10:10:11,1,北京市
  10. u001,A1,2019-09-02 10:10:11,1,北京市
  11. u001,A1,2019-09-02 10:10:11,1,北京市
  12. u001,A1,2019-09-02 10:10:11,1,北京市

运行结果如下： 发送的数据是活动id，Flink 通过活动id从MySQL获取活动名称

 

文章最后，给大家推荐一些受欢迎的技术博客链接：

  1. JAVA相关的深度技术博客链接
  2. Flink 相关技术博客链接
  3. Spark 核心技术链接
  4. 设计模式 —— 深度技术博客链接
  5. 机器学习 —— 深度技术博客链接
  6. Hadoop相关技术博客链接
  7. 超全干货--Flink思维导图，花了3周左右编写、校对
  8. 深入JAVA 的JVM核心原理解决线上各种故障【附案例】
  9. 请谈谈你对volatile的理解？--最近小李子与面试官的一场“硬核较量”
  10. 聊聊RPC通信，经常被问到的一道面试题。源码+笔记，包懂
  11. 深入聊聊Java 垃圾回收机制【附原理图及调优方法】

欢迎扫描下方的二维码或 搜索 公众号“大数据高级架构师”，我们会有更多、且及时的资料推送给您，欢迎多多交流！

                                           

       

 

注：本文转载自blog.csdn.net的不埋雷的探长的文章"https://blog.csdn.net/weixin_32265569/article/details/108367968"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！