最近有個(gè)需求砚作，需要把我們Structured Streaming處理后的實(shí)時(shí)數(shù)據(jù)，發(fā)送到Redis一份嘹锁。官網(wǎng)并沒有提供redis輸出方式葫录。之前我們使用的是foreachBatch這種方式，可以同時(shí)輸出到關(guān)系型數(shù)據(jù)庫领猾，kafka等米同，但是官方?jīng)]提供輸出方法的redis就有點(diǎn)難處理。后來看官方文檔摔竿，官方推薦我們使用foreach進(jìn)行輸出窍霞。對(duì)于我們這種需要往多個(gè)數(shù)據(jù)源同時(shí)輸出的情況，我們需要自定義Output Sink：
自定義sink需要繼承自ForeachWriter拯坟。以下是我寫的同時(shí)輸出到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
*
*
* 對(duì)于partition_id的每個(gè)分區(qū)：
*
* 對(duì)于epoch_id的流數(shù)據(jù)的每個(gè)批次/紀(jì)元：
*
* 方法open（partitionId但金，epochId）被調(diào)用。
*
* 如果open（...）返回true郁季，則對(duì)于分區(qū)和批處理/紀(jì)元中的每一行冷溃，將調(diào)用方法進(jìn)程（行）。
*
* 調(diào)用方法close（錯(cuò)誤）梦裂，在處理行時(shí)看到錯(cuò)誤（如果有）似枕。
*/
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中每一個(gè)字段
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()
//以下為發(fā)至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))
}
//記得關(guān)閉各種連接
override def close(errorOrNull: Throwable): Unit = {
//關(guān)閉連接
println("close connection !")
statementToInsert.executeBatch() //批量執(zhí)行
connection.commit //提交
//注意關(guān)閉各種連接
statementToInsert.close()
connection.close()
jedis.close()
}
}
主函數(shù)里面我們只需要如下調(diào)用即可：
val query = df.writeStream.outputMode("append").foreach(new MySink).start()
query.awaitTermination()
