Flink DataStream API实时计算结果写入MySQL数据库中并更新表中原主键对应的数据（scala）

scala代码，使用flink datastream计算的实时结果存入mysql数据库中

package com.test

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import java.sql.{Connection, DriverManager, PreparedStatement}



class MysqlSinkTumbling extends RichSinkFunction[Cal] {

  val driver = "com.mysql.jdbc.Driver"
  val url = "jdbc:mysql://192.168.xx.xxx:3306/database_name"
  val userName = "xxxx"
  val passWord = "xxxxxxxx"

  private var connection: Connection = null
  private var ps: PreparedStatement = null

  override def open(parameters: Configuration): Unit = {
    Class.forName(driver)
    connection = DriverManager.getConnection(url, userName, passWord)
    val sql = "insert into table_name(x1,x2,x3) values(?,?,?) on duplicate key update count=values(x3);"
    ps = connection.prepareStatement(sql)
  }

  override def invoke(value: Cal): Unit = {
    try {
      ps.setString(1, value.x1)
      ps.setString(2, value.x2)
      ps.setInt(3, value.x3)
      ps.executeUpdate()
    } catch {
      case e: Exception => println(e.getMessage)
    }
  }
  override def close(): Unit = {
    if(connection != null) {
      connection.close()
    }
    if(ps != null){
      ps.close()
    }
  }

}