方式一：JDBCInputFormat
1. 引入依賴
2. 通過TypeInformation創(chuàng)建 RowTypeInfo
3. 通過createInput 讀mysql表
4.通過 output寫mysql 表
方式二：自定義數(shù)據(jù)源
理論
SouceFunction 接口是所有stream source 的根接口，繼承自一個標記接口（空接口）Function并蝗。
在IDEA中打開SourceFunction卧檐，按下圖右擊鼠標，選擇Diagrams–>show Diagrams
SourceFunction 定義了兩個接口方法：
1忱辅、run：啟動一個source七蜘，即對接一個外部數(shù)據(jù)源讓然后emit元素形成stream（大部分情況下會通過在該方法里運行一個while循環(huán)的形式來產(chǎn)生stream）
2、cancel：取消一個source墙懂，即終止run中的循環(huán)emit元素的行為
正常情況下橡卤，一個SourceFunction實現(xiàn)這兩個接口方法就可以了。其實者兩個接口方法也固定了一種實現(xiàn)模板损搬。
實踐 - java版1
1. mysql數(shù)據(jù)庫建表碧库、插入數(shù)據(jù)
insert into user_order_count values ('16935394', 6), ('16374609', 4), ……;
2. pom.xml 中添加MySQL依賴
3. 新建對應的實體類 UserOrderCount
4. 新建Source 類 SourceFromMySQL
繼承RichSourceFunction，實現(xiàn) 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，實現(xiàn) open剩瓶、close驹溃、run、cancel 方法
2. 主程序
env.addSource(new mysql_source())
實踐 - scala版2
1.用case class 定義MySQL表
2. 實現(xiàn)RichSourceFunction來自定義MySQLSource
3. 主程序
4. 運行結果
兩種方式對比
- 如果設置多并行度延曙，從JDBCInput讀會采用設置的并行度讀數(shù)據(jù)豌鹤，會造成數(shù)據(jù)重復讀；而自定義數(shù)據(jù)源不會
- JDBCInput讀取的數(shù)據(jù)以ROW類型返回枝缔，寫入時也必須為ROW類型傍药，不方便使用pojo。JDBCInput返回的是DataSet，自定義數(shù)據(jù)源返回的是DataStreamSource
- JDBCInput需要為每個字段指定類型BasicTypeInfo拐辽，字段較多時不方便拣挪。
優(yōu)質資料搬運工：
https://www.pianshen.com/article/99991786993/ flink讀寫MySQL的兩種方式
https://blog.csdn.net/lz6363/article/details/114088147 【Flink入門】Flink自定義Source讀取MySQL數(shù)據(jù)
https://my.oschina.net/u/4228050/blog/4882932 flink 1.10 source方式讀取mysql
http://www.reibang.com/p/728d880dc59a Flink自定義MySQLSource讀取MySQL數(shù)據(jù)
