Flink实时统计入MySQL

Posted 2022-11-26 -早起的码农

先上代码片段

import java.sql.Types
import java.text.SimpleDateFormat
import java.util.Date

import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat
import org.apache.flink.api.scala._
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.DataStream, StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.types.Row


object WordCount 

   def main(args: Array[String]):Unit = 
     val port: Int = try 
       ParameterTool.fromArgs(args).getInt("port")
      catch 
       case e: Exception =>
         System.err.println("No port specified. Please run \'SocketWindowWordCount --port <port>\'")
         return
     

     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

     val text = env.socketTextStream("192.168.1.238", port, \'\\n\')


     val query