Structured Streaming自定义Output Sink到多个输出源

最近有个需求，需要把我们Structured Streaming处理后的实时数据，发送到Redis一份。官网并没有提供redis输出方式。之前我们使用的是foreachBatch这种方式，可以同时输出到关系型数据库，kafka等，但是官方没提供输出方法的redis就有点难处理。后来看官方文档，官方推荐我们使用foreach进行输出。对于我们这种需要往多个数据源同时输出的情况，我们需要自定义Output Sink：
自定义sink需要继承自ForeachWriter。以下是我写的同时输出到kafka,redis和mysql的sink类

package xds.DataCleaning_201905

import java.sql.{Connection, PreparedStatement}
import java.util
import java.util.Date
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.sql.{ForeachWriter, Row}
import org.json4s.jackson.JsonMethods.{compact, render}
import redis.clients.jedis.{Jedis}
import xds.Utils.{DateUtils, KafkaProducerUtils, MysqlManager, RedisClient}
import org.json4s._
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._

/**
  * wh 20190621
  *
  *
  * 对于partition_id的每个分区：
  *
  * 对于epoch_id的流数据的每个批次/纪元：
  *
  * 方法open（partitionId，epochId）被调用。
  *
  * 如果open（...）返回true，则对于分区和批处理/纪元中的每一行，将调用方法进程（行）。
  *
  * 调用方法close（错误），在处理行时看到错误（如果有）。
  */
class MySink extends ForeachWriter [Row]{
  val kafkaTopic : String = "LS_VD_CL"
  var jedis: Jedis = _
  var connection :Connection = _
  var statementToInsert : PreparedStatement = _
  var kafkaProducer : KafkaProducer[String, String] = _
  override def open(partitionId: Long, version: Long): Boolean = {
    jedis = RedisClient.pool.getResource
    connection = MysqlManager.getMysqlManager.getConnection
    kafkaProducer = KafkaProducerUtils.getProducer
    connection.setAutoCommit(false)
    statementToInsert = connection.prepareStatement("insert into t_videodata_1min (CreateTime,VehicleCount,Speed,ID_Link,ID_Station,ID_Lane,ID_TrafficSource,Type)" +
      "values (?,?,?,?,?,?,?,?)")
    println("open connection !")
    true
  }
  override def process(value: Row): Unit = {
    //获取row中每一个字段
    val CreateTime:Date = value.getAs[Date](0)
    val VehicleCount:Float = value.getAs[Float](1)
    val Speed:Float= value.getAs[Float](2)
    val ID_Link:String = value.getAs[String](3)
    val ID_Station:String = value.getAs[String](4)
    val ID_Lane:String = value.getAs[String](5)
    val ID_TrafficSource:String = value.getAs[String](6)
    val Type:Integer = value.getAs[Integer](7)
//以下为存入redis
    val map :util.HashMap[String,String]= new util.HashMap[String,String]
    map.put("VehicleCount",VehicleCount.toString)
    map.put("Speed",Speed.toString)
    map.put("ID_Lane",ID_Lane)
    val hourMin = DateUtils.dateToStr(CreateTime,"HHmm")
    jedis.hmset("C"+hourMin+ID_Link+"#"+ID_TrafficSource,map)
    val createTimeStr = DateUtils.dateToStr(CreateTime,"yyyy-MM-dd HH:mm:ss")
//以下为存入mysql
    statementToInsert.setObject(1,CreateTime)
    statementToInsert.setObject(2,VehicleCount)
    statementToInsert.setObject(3,Speed)
    statementToInsert.setObject(4,ID_Link)
    statementToInsert.setObject(5,ID_Station)
    statementToInsert.setObject(6,ID_Lane)
    statementToInsert.setObject(7,ID_TrafficSource)
    statementToInsert.setObject(8,Type)
    statementToInsert.addBatch()
//以下为发至kafka
    val messageToKafka = ("ID_TrafficSource" -> ID_TrafficSource) ~
      ("CreateTime" -> createTimeStr)~
      ("ID_Station" -> ID_Station) ~
      ("ID_Link" -> ID_Link)~
      ("ID_Lane" -> ID_Lane)~
      ("VehicleCount" -> VehicleCount) ~
      ("Speed" -> Speed) ~
      ("Type" -> Type.toString)
    val jsonToKafka = compact(render(messageToKafka))//封装成json
    kafkaProducer.send(new ProducerRecord(kafkaTopic,jsonToKafka))
  }
//记得关闭各种连接
  override def close(errorOrNull: Throwable): Unit = {
    //关闭连接
    println("close connection !")
    statementToInsert.executeBatch() //批量执行
    connection.commit //提交
    //注意关闭各种连接
    statementToInsert.close()
    connection.close()
    jedis.close()
  }
}

主函数里面我们只需要如下调用即可：

val query = df.writeStream.outputMode("append").foreach(new MySink).start()
query.awaitTermination()

最后编辑于：2019.06.29 14:24:21