1. 主程序

import java.util.Properties

import function.FileToMysqlWindowFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import sink.SinkToMysql
import source.SourceFromFile
import vo.ClassInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer


object FileToMysql {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    var props: Properties = new Properties();
    props.setProperty("bootstrap.servers", "localhost:9092");
    props.setProperty("group.id", "demoGroup");

    val kafkaConsumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("demoTopic", new SimpleStringSchema(), props);
    kafkaConsumer.setStartFromLatest();

    env
      .addSource(kafkaConsumer)
      .setParallelism(2)
      .map(line => parse(line))
      .timeWindowAll(Time.seconds(60))
      .process(new FileToMysqlWindowFunction)
      .addSink(new SinkToMysql)
    env.execute()
  }


  private def parse(line: String): ClassInfo = {
    var classInfo: ClassInfo = new ClassInfo()
    val arr = line.split(",")
    classInfo.id = arr(0)
    classInfo.classname = arr(1)
    classInfo.teacherId = arr(2)
    classInfo
  }

}

1. 利用 timeWindowAll 对 DataStream 画窗口
2. 利用 FileToMysqlWindowFunction 将窗口内的数据组成 List
3. 将 List 里的数据批量写入 MySQL

2. function

import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import vo.ClassInfo

import scala.collection.mutable.ListBuffer

class FileToMysqlWindowFunction extends ProcessAllWindowFunction[ClassInfo, List[ClassInfo], TimeWindow] {

  override def process(context: Context, elements: Iterable[ClassInfo], out: Collector[List[ClassInfo]]): Unit = {
    var buffer: ListBuffer[ClassInfo] = ListBuffer()
    val iterator = elements.iterator
    while (iterator.hasNext) {
      buffer.append(iterator.next())
    }

    out.collect(buffer.toList)
  }

}

3. Sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import vo.ClassInfo

class SinkToMysql extends RichSinkFunction[List[ClassInfo]] {

  private val jdbcUrl = "jdbc:mysql://localhost:3306/szw?useSSL=false&serverTimezone=GMT%2B8"
  private val username = "root"
  private val password = ""
  private val driverName = "com.mysql.cj.jdbc.Driver"
  private var connection: Connection = null
  private var ps: PreparedStatement = null

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)

    Class.forName(driverName)
    connection = DriverManager.getConnection(jdbcUrl, username, password)

    val sql: String = "replace into class_info(id, classname, teacher_id) values(?,?,?)"
    ps = connection.prepareStatement(sql)
  }

  override def invoke(list: List[ClassInfo], context: SinkFunction.Context[_]): Unit = {
    for (classInfo <- list) {
      ps.setString(1, classInfo.id)
      ps.setString(2, classInfo.classname)
      ps.setString(3, classInfo.teacherId)
      ps.addBatch()
    }
    ps.executeBatch()
  }

}