如何将spark streaming DF写入Kafka主题
How to write spark streaming DF to Kafka topic
我正在使用 Spark Streaming 处理两个 Kafka 队列之间的数据，但我似乎找不到从 Spark 写入 Kafka 的好方法。我试过这个：
input.foreachRDD(rdd =>
rdd.foreachPartition(partition =>
partition.foreach {
case x: String => {
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
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")
println(x)
val producer = new KafkaProducer[String, String](props)
val message = new ProducerRecord[String, String]("output", null, x)
producer.send(message)
}
}
)
)
它按预期工作，但在真实上下文中为每条消息实例化一个新的 KafkaProducer 显然是不可行的，我正在尝试解决它。
我想为每个进程保留对单个实例的引用，并在需要发送消息时访问它。如何从 Spark Streaming 写入 Kafka？
我的第一个建议是尝试在 foreachPartition 中创建一个新实例并测量它是否足够快以满足您的需求（官方文档建议在 foreachPartition 中实例化重对象）。
另一种选择是使用对象池，如本例所示：
然而我发现在使用检查点时很难实现。
另一个适合我的版本是工厂，如以下博客 post 所述，您只需检查它是否提供足够的并行度来满足您的需求（查看评论部分）：
为什么不可行？从根本上说，每个 RDD 的每个分区都将独立 运行（并且很可能 运行 在不同的集群节点上），因此您 必须 重做连接（并且任何同步）在每个分区的任务开始时。如果开销太高，那么您应该增加 StreamingContext 中的批处理大小，直到可以接受为止（当然，这样做会产生延迟成本）。
（如果您不在每个分区中处理数千条消息，您确定需要 spark-streaming 吗？使用独立应用程序会更好吗？）
有一个由 Cloudera 维护的 Streaming Kafka Writer（实际上是从 Spark JIRA [1] 中分离出来的）。它基本上为每个分区创建了一个生产者，它分摊了在（希望是大的）元素集合上创建 'heavy' 对象所花费的时间。
这可能是您想要做的。您基本上为每个记录分区创建一个生产者。
input.foreachRDD(rdd =>
rdd.foreachPartition(
partitionOfRecords =>
{
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
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)
partitionOfRecords.foreach
{
case x:String=>{
println(x)
val message=new ProducerRecord[String, String]("output",null,x)
producer.send(message)
}
}
})
)
希望对您有所帮助
我遇到了同样的问题并找到了 this post。
作者通过为每个执行者创建1个生产者来解决问题。他没有发送生产者本身，而是仅发送了一个“配方”，即如何通过广播在执行者中创建生产者。
val kafkaSink = sparkContext.broadcast(KafkaSink(conf))
他使用了一个懒惰地创建生产者的包装器：
class KafkaSink(createProducer: () => KafkaProducer[String, String]) extends Serializable {
lazy val producer = createProducer()
def send(topic: String, value: String): Unit = producer.send(new ProducerRecord(topic, value))
}
object KafkaSink {
def apply(config: Map[String, Object]): KafkaSink = {
val f = () => {
val producer = new KafkaProducer[String, String](config)
sys.addShutdownHook {
producer.close()
}
producer
}
new KafkaSink(f)
}
}
包装器是可序列化的，因为 Kafka 生产者在首次用于执行器之前被初始化。驱动程序保留对包装器的引用，包装器使用每个执行程序的生产者发送消息：
dstream.foreachRDD { rdd =>
rdd.foreach { message =>
kafkaSink.value.send("topicName", message)
}
}
是的，不幸的是，Spark（1.x、2.x）并没有直接说明如何以有效的方式写入 Kafka。
我建议采用以下方法：
- 每个执行者使用（和重复使用）一个
KafkaProducer 个实例 process/JVM。
这是此方法的高级设置：
- 首先，你必须 "wrap" Kafka 的
KafkaProducer 因为，正如你提到的，它是不可序列化的。包装它可以让你 "ship" 它给执行者。这里的关键思想是使用 lazy val，这样你就可以延迟实例化生产者直到它第一次使用，这实际上是一种解决方法，这样你就不必担心 KafkaProducer 无法序列化。
- 您 "ship" 通过使用广播变量将包装的生产者发送给每个执行者。
- 在您的实际处理逻辑中，您通过广播变量访问包装的生产者，并使用它将处理结果写回 Kafka。
从 Spark 2.0 开始，以下代码片段可与 Spark Streaming 一起使用。
第 1 步：包装 KafkaProducer
import java.util.concurrent.Future
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
class MySparkKafkaProducer[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
/* This is the key idea that allows us to work around running into
NotSerializableExceptions. */
lazy val producer = createProducer()
def send(topic: String, key: K, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, key, value))
def send(topic: String, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, value))
}
object MySparkKafkaProducer {
import scala.collection.JavaConversions._
def apply[K, V](config: Map[String, Object]): MySparkKafkaProducer[K, V] = {
val createProducerFunc = () => {
val producer = new KafkaProducer[K, V](config)
sys.addShutdownHook {
// Ensure that, on executor JVM shutdown, the Kafka producer sends
// any buffered messages to Kafka before shutting down.
producer.close()
}
producer
}
new MySparkKafkaProducer(createProducerFunc)
}
def apply[K, V](config: java.util.Properties): MySparkKafkaProducer[K, V] = apply(config.toMap)
}
第 2 步：使用广播变量为每个执行程序提供自己包装的 KafkaProducer 实例
import org.apache.kafka.clients.producer.ProducerConfig
val ssc: StreamingContext = {
val sparkConf = new SparkConf().setAppName("spark-streaming-kafka-example").setMaster("local[2]")
new StreamingContext(sparkConf, Seconds(1))
}
ssc.checkpoint("checkpoint-directory")
val kafkaProducer: Broadcast[MySparkKafkaProducer[Array[Byte], String]] = {
val kafkaProducerConfig = {
val p = new Properties()
p.setProperty("bootstrap.servers", "broker1:9092")
p.setProperty("key.serializer", classOf[ByteArraySerializer].getName)
p.setProperty("value.serializer", classOf[StringSerializer].getName)
p
}
ssc.sparkContext.broadcast(MySparkKafkaProducer[Array[Byte], String](kafkaProducerConfig))
}
第 3 步：从 Spark Streaming 写入 Kafka，重新使用相同的包装 KafkaProducer 实例（对于每个执行程序）
import java.util.concurrent.Future
import org.apache.kafka.clients.producer.RecordMetadata
val stream: DStream[String] = ???
stream.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
val metadata: Stream[Future[RecordMetadata]] = partitionOfRecords.map { record =>
kafkaProducer.value.send("my-output-topic", record)
}.toStream
metadata.foreach { metadata => metadata.get() }
}
}
希望这对您有所帮助。
With Spark >= 2.2
Both read and write operations are possible on Kafka using Structured Streaming API
从 Kafka 主题构建流
// Subscribe to a topic and read messages from the earliest to latest offsets
val ds= spark
.readStream // use `read` for batch, like DataFrame
.format("kafka")
.option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")
.option("subscribe", "source-topic1")
.option("startingOffsets", "earliest")
.option("endingOffsets", "latest")
.load()
读取键和值并为两者应用模式，为简单起见，我们将它们都转换为 String 类型。
val dsStruc = ds.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]
因为 dsStruc 有模式，它接受所有 SQL 类型的操作，如 filter、agg、select ..等。
将流写入 Kafka 主题
dsStruc
.writeStream // use `write` for batch, like DataFrame
.format("kafka")
.option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")
.option("topic", "target-topic1")
.start()
更多configuration for Kafka integration to read or write
要在应用程序中添加的关键工件
"org.apache.spark" % "spark-core_2.11" % 2.2.0,
"org.apache.spark" % "spark-streaming_2.11" % 2.2.0,
"org.apache.spark" % "spark-sql-kafka-0-10_2.11" % 2.2.0,
With Spark < 2.2
Since there is no direct way of writing the messages to Kafka from Spark Streaming
创建一个 KafkaSinkWritter
import java.util.Properties
import 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", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
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()
}
}
使用 SinkWriter 写入消息
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()
引用link
我正在使用 Spark Streaming 处理两个 Kafka 队列之间的数据，但我似乎找不到从 Spark 写入 Kafka 的好方法。我试过这个：
input.foreachRDD(rdd =>
rdd.foreachPartition(partition =>
partition.foreach {
case x: String => {
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
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")
println(x)
val producer = new KafkaProducer[String, String](props)
val message = new ProducerRecord[String, String]("output", null, x)
producer.send(message)
}
}
)
)
它按预期工作，但在真实上下文中为每条消息实例化一个新的 KafkaProducer 显然是不可行的，我正在尝试解决它。
我想为每个进程保留对单个实例的引用，并在需要发送消息时访问它。如何从 Spark Streaming 写入 Kafka？
我的第一个建议是尝试在 foreachPartition 中创建一个新实例并测量它是否足够快以满足您的需求（官方文档建议在 foreachPartition 中实例化重对象）。
另一种选择是使用对象池，如本例所示：
然而我发现在使用检查点时很难实现。
另一个适合我的版本是工厂，如以下博客 post 所述，您只需检查它是否提供足够的并行度来满足您的需求（查看评论部分）：
为什么不可行？从根本上说，每个 RDD 的每个分区都将独立 运行（并且很可能 运行 在不同的集群节点上），因此您 必须 重做连接（并且任何同步）在每个分区的任务开始时。如果开销太高，那么您应该增加 StreamingContext 中的批处理大小，直到可以接受为止（当然，这样做会产生延迟成本）。
（如果您不在每个分区中处理数千条消息，您确定需要 spark-streaming 吗？使用独立应用程序会更好吗？）
有一个由 Cloudera 维护的 Streaming Kafka Writer（实际上是从 Spark JIRA [1] 中分离出来的）。它基本上为每个分区创建了一个生产者，它分摊了在（希望是大的）元素集合上创建 'heavy' 对象所花费的时间。
这可能是您想要做的。您基本上为每个记录分区创建一个生产者。
input.foreachRDD(rdd =>
rdd.foreachPartition(
partitionOfRecords =>
{
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
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)
partitionOfRecords.foreach
{
case x:String=>{
println(x)
val message=new ProducerRecord[String, String]("output",null,x)
producer.send(message)
}
}
})
)
希望对您有所帮助
我遇到了同样的问题并找到了 this post。
作者通过为每个执行者创建1个生产者来解决问题。他没有发送生产者本身，而是仅发送了一个“配方”，即如何通过广播在执行者中创建生产者。
val kafkaSink = sparkContext.broadcast(KafkaSink(conf))
他使用了一个懒惰地创建生产者的包装器：
class KafkaSink(createProducer: () => KafkaProducer[String, String]) extends Serializable {
lazy val producer = createProducer()
def send(topic: String, value: String): Unit = producer.send(new ProducerRecord(topic, value))
}
object KafkaSink {
def apply(config: Map[String, Object]): KafkaSink = {
val f = () => {
val producer = new KafkaProducer[String, String](config)
sys.addShutdownHook {
producer.close()
}
producer
}
new KafkaSink(f)
}
}
包装器是可序列化的，因为 Kafka 生产者在首次用于执行器之前被初始化。驱动程序保留对包装器的引用，包装器使用每个执行程序的生产者发送消息：
dstream.foreachRDD { rdd =>
rdd.foreach { message =>
kafkaSink.value.send("topicName", message)
}
}
是的，不幸的是，Spark（1.x、2.x）并没有直接说明如何以有效的方式写入 Kafka。
我建议采用以下方法：
- 每个执行者使用（和重复使用）一个
KafkaProducer个实例 process/JVM。
这是此方法的高级设置：
- 首先，你必须 "wrap" Kafka 的
KafkaProducer因为，正如你提到的，它是不可序列化的。包装它可以让你 "ship" 它给执行者。这里的关键思想是使用lazy val，这样你就可以延迟实例化生产者直到它第一次使用，这实际上是一种解决方法，这样你就不必担心KafkaProducer无法序列化。 - 您 "ship" 通过使用广播变量将包装的生产者发送给每个执行者。
- 在您的实际处理逻辑中，您通过广播变量访问包装的生产者，并使用它将处理结果写回 Kafka。
从 Spark 2.0 开始，以下代码片段可与 Spark Streaming 一起使用。
第 1 步：包装 KafkaProducer
import java.util.concurrent.Future
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
class MySparkKafkaProducer[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
/* This is the key idea that allows us to work around running into
NotSerializableExceptions. */
lazy val producer = createProducer()
def send(topic: String, key: K, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, key, value))
def send(topic: String, value: V): Future[RecordMetadata] =
producer.send(new ProducerRecord[K, V](topic, value))
}
object MySparkKafkaProducer {
import scala.collection.JavaConversions._
def apply[K, V](config: Map[String, Object]): MySparkKafkaProducer[K, V] = {
val createProducerFunc = () => {
val producer = new KafkaProducer[K, V](config)
sys.addShutdownHook {
// Ensure that, on executor JVM shutdown, the Kafka producer sends
// any buffered messages to Kafka before shutting down.
producer.close()
}
producer
}
new MySparkKafkaProducer(createProducerFunc)
}
def apply[K, V](config: java.util.Properties): MySparkKafkaProducer[K, V] = apply(config.toMap)
}
第 2 步：使用广播变量为每个执行程序提供自己包装的 KafkaProducer 实例
import org.apache.kafka.clients.producer.ProducerConfig
val ssc: StreamingContext = {
val sparkConf = new SparkConf().setAppName("spark-streaming-kafka-example").setMaster("local[2]")
new StreamingContext(sparkConf, Seconds(1))
}
ssc.checkpoint("checkpoint-directory")
val kafkaProducer: Broadcast[MySparkKafkaProducer[Array[Byte], String]] = {
val kafkaProducerConfig = {
val p = new Properties()
p.setProperty("bootstrap.servers", "broker1:9092")
p.setProperty("key.serializer", classOf[ByteArraySerializer].getName)
p.setProperty("value.serializer", classOf[StringSerializer].getName)
p
}
ssc.sparkContext.broadcast(MySparkKafkaProducer[Array[Byte], String](kafkaProducerConfig))
}
第 3 步：从 Spark Streaming 写入 Kafka，重新使用相同的包装 KafkaProducer 实例（对于每个执行程序）
import java.util.concurrent.Future
import org.apache.kafka.clients.producer.RecordMetadata
val stream: DStream[String] = ???
stream.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
val metadata: Stream[Future[RecordMetadata]] = partitionOfRecords.map { record =>
kafkaProducer.value.send("my-output-topic", record)
}.toStream
metadata.foreach { metadata => metadata.get() }
}
}
希望这对您有所帮助。
With Spark >= 2.2
Both read and write operations are possible on Kafka using Structured Streaming API
从 Kafka 主题构建流
// Subscribe to a topic and read messages from the earliest to latest offsets
val ds= spark
.readStream // use `read` for batch, like DataFrame
.format("kafka")
.option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")
.option("subscribe", "source-topic1")
.option("startingOffsets", "earliest")
.option("endingOffsets", "latest")
.load()
读取键和值并为两者应用模式，为简单起见，我们将它们都转换为 String 类型。
val dsStruc = ds.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]
因为 dsStruc 有模式，它接受所有 SQL 类型的操作，如 filter、agg、select ..等。
将流写入 Kafka 主题
dsStruc
.writeStream // use `write` for batch, like DataFrame
.format("kafka")
.option("kafka.bootstrap.servers", "brokerhost1:port1,brokerhost2:port2")
.option("topic", "target-topic1")
.start()
更多configuration for Kafka integration to read or write
要在应用程序中添加的关键工件
"org.apache.spark" % "spark-core_2.11" % 2.2.0,
"org.apache.spark" % "spark-streaming_2.11" % 2.2.0,
"org.apache.spark" % "spark-sql-kafka-0-10_2.11" % 2.2.0,
With Spark < 2.2
Since there is no direct way of writing the messages to Kafka from Spark Streaming
创建一个 KafkaSinkWritter
import java.util.Properties
import 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", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
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()
}
}
使用 SinkWriter 写入消息
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()
引用link