導入依賴
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.46</version>
</dependency>
編譯代碼
package com.wyh.streamingApi.sink import java.sql.{Connection, DriverManager, PreparedStatement} import org.apache.flink.configuration.Configuration import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction} import org.apache.flink.streaming.api.scala._ object Sink2JDBC { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) //Source操作
val inputStream = env.readTextFile("F:\\flink-study\\wyhFlinkSD\\data\\sensor.txt") //Transform操作
val dataStream: DataStream[SensorReading] = inputStream.map(data => { val dataArray = data.split(",") SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble) }) //Sink操作
dataStream.addSink(new MyJDBCSink()) env.execute("JDBC sink Test") } } class MyJDBCSink() extends RichSinkFunction[SensorReading]{ //連接、定義預編譯器
var conn:Connection = _ var insertStmt:PreparedStatement = _ var updateStmt:PreparedStatement = _ //初始化 建立
override def open(parameters: Configuration): Unit = { super.open(parameters) conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test","root","root") insertStmt = conn.prepareStatement("insert into temperatures (sensor,temp) values (?,?)") updateStmt = conn.prepareStatement("update temperatures set temp = ? where sensor = ?") } //調用連接執行sql
override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = { //執行更新語句
updateStmt.setDouble(1,value.temperature) updateStmt.setString(2,value.id) updateStmt.execute() //如果update沒有查到數據，執行插入語句
if(updateStmt.getUpdateCount == 0){ insertStmt.setString(1,value.id) insertStmt.setDouble(2,value.temperature) insertStmt.execute() } } //關閉時 做清理工作
override def close(): Unit = { conn.close() insertStmt.close() updateStmt.close() } }
運行
