import org.apache.spark.sql.ForeachWriter
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature
object KafkaToPostgresql {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder
.appName(“kafka to postgresql”)
.master(“local[2]”)
.getOrCreate()
import spark.implicits._
val df = spark
.readStream
.format(“kafka”)
.option(“kafka.bootstrap.servers”, “xx.xx.xx.xx:9092”)
.option(“subscribe”, “topic-name”)
.option(“startingOffsets”,“latest”)
.load()
// 显示kafka报文格式
df.printSchema()
val rowDataset = df.selectExpr(“CAST(value AS STRING)”)
rowDataset.writeStream.foreach(new ForeachWriterRow{
override def process(record: Row): Unit = {
val mapper = new ObjectMapper()
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, fals 《一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》无偿开源 威信搜索公众号【编程进阶路】 e)
// 很重要
mapper.registerModule(DefaultScalaModule)
