flinkaddsource

class MySOurceMysql extends RichSourceFunction[String]{
  var conn: Connection =null
  var ps: PreparedStatement =null


  override def open(parameters: Configuration): Unit ={
  //  Driver
    Class.forName("com.mysql.jdbc.Driver")
   conn= DriverManager.getConnection("jdbc:mysql://hdp1:3306/1912b", "root", "root")
   ps=  conn.prepareStatement("select * from t_zk1")
  }

  override def close(): Unit ={
    conn.close()
    ps.close()
  }

  override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
    val rs: ResultSet = ps.executeQuery()
    while (rs.next()){
      val s1: String = rs.getString(1)
      val s2: String = rs.getString(2)
      sourceContext.collect(s1+","+s2)
    }
  }

  override def cancel(): Unit = ???
}