在前一篇博客中 【Flink入门】Flink读取Kafka数据Demo 已经简单介绍了Flink读取Kafka数据并通过Demo进行实践，这篇博客简单介绍Flink通过自定义Source读取MySQL数据并通过Demo进行演练。
首先我们来简单了解下SourceFunction 接口，它是所有 stream source 的根接口，它继承自一个标记接口（空接口）Function。
在IDEA中打开SourceFunction，按下图右击鼠标，选择Diagrams–>show Diagrams
SourceFunction 定义了两个接口方法：
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")
}
}
运行结果：
