Flink DataStream API （十四）Flink 输出到 MySQL（JDBC）

尽管在大数据处理中直接与 MySQL 交互的场景不多，但最终处理的计算结果是要给外部应用消费使用的，而外部应用读取的数据存储往往就是 MySQL。所以我们也需要知道如何将数据输出到 MySQL 这样的传统数据库。

写入数据的 MySQL 的测试步骤如下。
（1）添加依赖

    <!--Flink 对接 MySQL 依赖-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-jdbc_${scala.version}</artifactId>
        <version>1.11.0</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.47</version>
    </dependency>

（2）启动 MySQL，在 database 库下建表 clicks

CREATE TABLE `writetodb` (
  `sensorname` varchar(23) DEFAULT NULL,
  `timestamp` bigint(20) DEFAULT NULL,
  `wendu` double DEFAULT NULL
)

（3）编写输出到 MySQL 的示例代码

object TestToMySQLEasy {
    
	  def main(args: Array[String]): Unit = {
    
		    val env = StreamExecutionEnvironment.getExecutionEnvironment
		// env.enableCheckpointing(3000L) //毫秒，写入 Mysql 必须开启检查点
		    env.setParallelism(1)
		
		    val sql = "INSERT INTO writetodb (sensorname, timestamp, wendu) VALUE (?,?,?)"
		
		    env.readTextFile("/home/rjxy/zlp/Code/CodePro/GuoSai/Task01/src/main/resources/test.txt")
		      .map(elem => {
    
		        val data = elem.split(",")
		        (data(0), data(1).toLong, data(2).toDouble)
		      })
		      .addSink(JdbcSink.sink(
		        sql, //
		        new JdbcStatementBuilder[(String,Long,Double)] {
    
		          override def accept(ps: PreparedStatement, u: (String, Long, Double)): Unit = {
    
		            ps.setString(1, u._1);
		            ps.setLong(2, u._2);
		            ps.setDouble(3, u._3);
		          }
		        },
		        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
		          .withDriverName("com.mysql.jdbc.Driver")
		          .withUrl("jdbc:mysql://192.168.23.34:3306/zlp")
		          .withUsername("root")
		          .withPassword("123456")
		          .build()
		      ));
		
		    env.execute()
	  }
}

（4）运行代码，用客户端连接 MySQL，查看是否成功写入数据。