spark streaming-从kafka读取json并将json写入其他kafka主题

ctehm74n 于 2021-06-07 发布在 Kafka

关注(0)|答案(2)|浏览(369)

我正在准备spark流媒体的应用程序（spark 2.1，kafka 0.10）
我需要从Kafka主题“输入”中读取数据，找到正确的数据并将结果写入主题“输出”
我可以基于kafkautils.createdirectstream方法从kafka读取数据。
我将rdd转换为json并准备过滤器：

val messages = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)
val elementDstream = messages.map(v => v.value).foreachRDD { rdd =>
  val PeopleDf=spark.read.schema(schema1).json(rdd)
  import spark.implicits._
  PeopleDf.show()
  val PeopleDfFilter = PeopleDf.filter(($"value1".rlike("1"))||($"value2" === 2))
  PeopleDfFilter.show()
}

我可以从kafka加载数据，并使用kafkaproducer将“原样”写入kafka：

messages.foreachRDD( rdd => {
      rdd.foreachPartition( partition => {
        val kafkaTopic = "output"
        val props = new HashMap[String, Object]()
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringSerializer")
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringSerializer")
        val producer = new KafkaProducer[String, String](props)
        partition.foreach{ record: ConsumerRecord[String, String] => {
        System.out.print("########################" + record.value())
        val messageResult = new ProducerRecord[String, String](kafkaTopic, record.value())
        producer.send(messageResult)
        }}
        producer.close()
      })
    })

但我不能将这两个操作集成到json中，即在json中找到合适的值并将结果写入kafka：以json格式编写peopledfilter以“输出”kafka主题。
我在kafka中有很多输入消息，这就是为什么我要使用foreachpartition来创建kafka producer的原因。
非常感谢你的建议。

scala apache-kafka apache-spark spark-streaming

来源：https://stackoverflow.com/questions/47457755/spark-streaming-read-json-from-kafka-and-write-json-to-other-kafka-topic

2条答案

按热度按时间

cu6pst1q1#

尝试使用结构化流媒体。即使您使用了spark 2.1，也可以实现您自己的kafka foreachwriter，如下所示：
KafkaFlume：

import java.util.Properties
import kafkashaded.org.apache.kafka.clients.producer._
import org.apache.spark.sql.ForeachWriter
 class  KafkaSink(topic:String, servers:String) extends ForeachWriter[(String, String)] {
      val kafkaProperties = new Properties()
      kafkaProperties.put("bootstrap.servers", servers)
      kafkaProperties.put("key.serializer",
        classOf[org.apache.kafka.common.serialization.StringSerializer].toString)
      kafkaProperties.put("value.serializer",
        classOf[org.apache.kafka.common.serialization.StringSerializer].toString)
      val results = new scala.collection.mutable.HashMap[String, String]
      var producer: KafkaProducer[String, String] = _
      def open(partitionId: Long,version: Long): Boolean = {
        producer = new KafkaProducer(kafkaProperties)
        true
      }
      def process(value: (String, String)): Unit = {
          producer.send(new ProducerRecord(topic, value._1 + ":" + value._2))
      }
      def close(errorOrNull: Throwable): Unit = {
        producer.close()
      }
   }

用法：

val topic = "<topic2>"
val brokers = "<server:ip>"
val writer = new KafkaSink(topic, brokers)
val query =
  streamingSelectDF
    .writeStream
    .foreach(writer)
    .outputMode("update")
    .trigger(ProcessingTime("25 seconds"))
    .start()

展开查看全部

赞(0）回复(0）举报 2021-06-07

sdnqo3pr2#

这个过程非常简单，为什么不一直使用结构化流媒体呢？

import org.apache.spark.sql.functions.from_json
spark
  // Read the data
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", inservers) 
  .option("subscribe", intopic)
  .load()
  // Transform / filter
  .select(from_json($"value".cast("string"), schema).alias("value"))
  .filter(...)  // Add the condition
  .select(to_json($"value").alias("value")
  // Write back
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", outservers)
  .option("subscribe", outtopic)
  .start()

展开查看全部

赞(0）回复(0）举报 2021-06-07

我来回答

spark streaming-从kafka读取json并将json写入其他kafka主题

2条答案

相关问题

热门标签

最新问答