方式一：JDBCInputFormat
1. 引入依赖
2. 通过TypeInformation创建 RowTypeInfo
3. 通过createInput 读mysql表
4.通过 output写mysql 表
方式二：自定义数据源
理论
SouceFunction 接口是所有stream source 的根接口，继承自一个标记接口（空接口）Function。
在IDEA中打开SourceFunction，按下图右击鼠标，选择Diagrams–>show Diagrams
SourceFunction 定义了两个接口方法：
1、run：启动一个source，即对接一个外部数据源让然后emit元素形成stream（大部分情况下会通过在该方法里运行一个while循环的形式来产生stream）
2、cancel：取消一个source，即终止run中的循环emit元素的行为
正常情况下，一个SourceFunction实现这两个接口方法就可以了。其实者两个接口方法也固定了一种实现模板。
实践 - java版1
1. mysql数据库建表、插入数据
insert into user_order_count values ('16935394', 6), ('16374609', 4), ……;
2. pom.xml 中添加MySQL依赖
3. 新建对应的实体类 UserOrderCount
4. 新建Source 类 SourceFromMySQL
继承RichSourceFunction，实现 open、close、run、cancel方法
con = DriverManager.getConnection("jdbc:mysql://host:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
System.out.println("-----------mysql get connection has exception, msg = " + e.getMessage());
5. flink 主程序
6. 运行结果
实践 - java版2
sourceFromMySQL类（代码里写的MysqlSource，同java版1中的sourceFromMySQL）
2. 主程序
实践 - scala版1
1. sourceFromMySQL类(代码里写的mysql_source，同java方法中的sourceFromMySQL)
继承RichSourceFuncion，实现 open、close、run、cancel 方法
2. 主程序
env.addSource(new mysql_source())
实践 - scala版2
1.用case class 定义MySQL表
2. 实现RichSourceFunction来自定义MySQLSource
3. 主程序
4. 运行结果
两种方式对比
- 如果设置多并行度，从JDBCInput读会采用设置的并行度读数据，会造成数据重复读；而自定义数据源不会
- JDBCInput读取的数据以ROW类型返回，写入时也必须为ROW类型，不方便使用pojo。JDBCInput返回的是DataSet，自定义数据源返回的是DataStreamSource
- JDBCInput需要为每个字段指定类型BasicTypeInfo，字段较多时不方便。
优质资料搬运工：
https://www.pianshen.com/article/99991786993/ flink读写MySQL的两种方式
https://blog.csdn.net/lz6363/article/details/114088147 【Flink入门】Flink自定义Source读取MySQL数据
https://my.oschina.net/u/4228050/blog/4882932 flink 1.10 source方式读取mysql
https://www.jianshu.com/p/728d880dc59a Flink自定义MySQLSource读取MySQL数据
