Flink 寫數據到MySql (JDBC Sink)

POM 文件

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.11</artifactId>
            <version>1.10.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>1.10.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
            <version>1.10.1</version>
        </dependency>
 <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.44</version> </dependency>

代碼:

package com.kpwong.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.{DataStream, StreamExecutionEnvironment} import org.apache.flink.streaming.api.scala._ object JDBCSinkTest { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) val socketDS: DataStream[String] = env.socketTextStream("hadoop202",9999) //sink
 socketDS.addSink(new MyJDBCSink()) env.execute() } } //注意 必須繼承富函數類,這樣才有生命周期管理函數 class MyJDBCSink extends RichSinkFunction[String]{ //定義SQL 鏈接 預編譯器
  var conn: Connection =_ var insertSql: PreparedStatement=_ //創建鏈接和預編譯語句
  override def open(parameters: Configuration): Unit = { super.open(parameters) conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test","root","000000") insertSql= conn.prepareStatement("insert into flink(socket) values (?)") } override def invoke(value: String, context: SinkFunction.Context[_]): Unit = { // super.invoke(value, context)
    insertSql.setString(1,value) insertSql.execute() } override def close(): Unit = { // super.close()
 insertSql.close() conn.close() } }

測試 ：socket輸入輸入數據

 

 MySql 接受數據: