【Flink入门】Flink自定义Source读取MySQL数据

╰半夏微凉° 2021-09-25 04:44 872阅读 0赞

在前一篇博客中 [【Flink入门】Flink读取Kafka数据Demo][Flink_Flink_Kafka_Demo] 已经简单介绍了Flink读取Kafka数据并通过Demo进行实践，这篇博客简单介绍Flink通过自定义Source读取MySQL数据并通过Demo进行演练。

首先我们来简单了解下SourceFunction 接口，它是所有 stream source 的根接口，它继承自一个标记接口（空接口）Function。

在IDEA中打开SourceFunction，按下图右击鼠标，选择Diagrams–>show Diagrams  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70]

SourceFunction 定义了两个接口方法：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 1]

1、run ： 启动一个 source，即对接一个外部数据源然后 emit 元素形成 stream（大部分情况下会通过在该方法里运行一个 while 循环的形式来产生 stream）。  
2、cancel ： 取消一个 source，也即将 run 中的循环 emit 元素的行为终止。  
正常情况下，一个 SourceFunction 实现这两个接口方法就可以了。其实这两个接口方法也固定了一种实现模板。

接下来通过Demo实现  
首先 pom.xml 中添加 MySQL 依赖：

<dependency>
         <groupId>mysql</groupId>
         <artifactId>mysql-connector-java</artifactId>
         <version>5.1.27</version>
    </dependency>

MySQL数据库建表

drop table if exists user_order_count;
    create table user_order_count (
    user_id varchar(25) NOT NULL,
    count int(11),
    primary key (user_id)
    ) engine=innodb default charset=utf8 collate=utf8_bin;

导入模拟数据

insert into user_order_count values ('16935394', 6), ('16374609', 4), ('16570065', 4), ('4611433', 3), ('17308713', 3);

新建对应的实体类：UserOrderCount

package com.fuyun.flink.model;
    
    public class UserOrderCount { 
        public String userId;
        public int count;
    
        public UserOrderCount() { 
        }
    
        public UserOrderCount(String userId, int count){ 
            this.userId = userId;
            this.count = count;
        }
        @Override
        public String toString() { 
            return "UserOrderCount{" +
                    "userId=" + userId +
                    ", count=" + count +
                    '}';
        }
    
        public String getUserId() { 
            return userId;
        }
    
        public void setUserId(String userId) { 
            this.userId = userId;
        }
    
        public int getCount() { 
            return count;
        }
    
        public void setCount(int count) { 
            this.count = count;
        }
    }

新建 Source 类 SourceFromMySQL.java，该类继承 RichSourceFunction ，实现里面的 open、close、run、cancel 方法：

package com.fuyun.flink.souce;
    
    import com.fuyun.flink.model.UserOrderCount;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    import java.sql.ResultSet;
    
    public class SourceFromMySQL extends RichSourceFunction<UserOrderCount> { 
        PreparedStatement ps;
        private Connection connection;
    
        /** * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。 * * @param parameters * @throws Exception */
        @Override
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            connection = getConnection();
            String sql = "select * from user_order_count;"; // 编写具体逻辑代码
            ps = this.connection.prepareStatement(sql);
        }
    
        /** * 程序执行完毕就可以进行，关闭连接和释放资源的动作了 * * @throws Exception */
        @Override
        public void close() throws Exception { 
            super.close();
            if (connection != null) {  //关闭连接和释放资源
                connection.close();
            }
            if (ps != null) { 
                ps.close();
            }
        }
    
        @Override
        public void run(SourceContext<UserOrderCount> ctx) throws Exception { 
            ResultSet resultSet = ps.executeQuery(); // 执行SQL语句返回结果集
            while (resultSet.next()) { 
                UserOrderCount userOrderCount = new UserOrderCount(
                        resultSet.getString("user_id").trim(),
                        resultSet.getInt("count"));
                ctx.collect(userOrderCount);
            }
        }
    
        @Override
        public void cancel() { 
        }
    
        private static Connection getConnection() { 
            Connection con = null;
            try { 
                Class.forName("com.mysql.jdbc.Driver");
                con = DriverManager.getConnection("jdbc:mysql://bigdata-training.fuyun.com:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
            } catch (Exception e) { 
                System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());
            }
            return con;
        }
    }

Flink主程序

package com.fuyun.flink
    
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
    import com.fuyun.flink.souce.SourceFromMySQL
    
    object SourceMain { 
      def main(args: Array[String]): Unit = { 
        // 创建流处理环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment()
    
        env.addSource(new SourceFromMySQL).print
    
        env.execute("Flink add data sourc")
      }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 2]

[Flink_Flink_Kafka_Demo]: https://blog.csdn.net/lz6363/article/details/113994011
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70]: /images/20210923/58b3f681f3f94a66a2ecb8bf9f6fd6f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 1]: /images/20210923/5efadf9e53dd45e7a7b908b07a5a1093.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 2]: /images/20210923/41a3e025ce974251acb5dbb91603b96b.png