Flink用RichParallelSourceFunction连接mysql

import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.slf4j.LoggerFactory

import scala.collection.mutable


class SouceParaMysql(env:StreamExecutionEnvironment) extends RichParallelSourceFunction[mutable.HashMap[String,String]]{
  val logger = LoggerFactory.getLogger("SouceParaMysql")
  var conn:Connection = _
  var selectStnt:PreparedStatement = _
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    val driver = "com.mysql.cj.jdbc.Driver"
    Class.forName(driver).isInstance()
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/flink","root","111111")

  }

  /**
    *
    * @param ctx
    */
  override def run(ctx: SourceFunction.SourceContext[mutable.HashMap[String, String]]): Unit = {
    selectStnt = conn.prepareStatement("select.......")
    val sql_result = selectStnt.executeQuery
    var data = mutable.HashMap[String,String]()
    while(sql_result.next()){
      val ids = sql_result.getString(1).trim()
      val domain = sql_result.getString(2).trim()
      data.put(domain,ids)
    }
    ctx.collect(data)
  }

  override def cancel(): Unit = {
    if (selectStnt != null){
      selectStnt.close()
    }
    if (conn != null){
      conn.close()
    }
  }


}

之前有踩坑，一直出现异常关闭的错误。
原因是
selectStnt = conn.prepareStatement(“select…”)
同时改为class
这条语句写在open内，具体原因未知，但是这样改是可以的。