在单个生产命令中可以为 Kafka 主题生成的记录数量是否有任何限制答案

【问题标题】：Is there any limit to number of records that can be produced to a Kafka topic in single produce command在单个生产命令中可以为 Kafka 主题生成的记录数量是否有任何限制
【发布时间】：2020-12-18 16:19:46
【问题描述】：

我有一个 Databricks Kafka Producer，需要将 62M 记录写入 Kafka 主题。如果我同时写62M条记录会不会有问题？还是我需要迭代 20 次，每次迭代写入 3M 条记录。

这里是代码。

Cmd1 val srcDf = spark.read.format("delta").load("/mnt/data-lake/data/silver/geocodes").filter($"LastUpdateDt"===lastUpdateDt)

Cmd2 val strDf = srcDf
        .withColumn("key",...
        .withColumn("topLevelRecord",...

Cmd3 strDf
 .select(
 to_avro($"key", lit("topic-AVRO-key"), schemaRegistryAddr).as("key"),
 to_avro($"topLevelRecord", lit("topic-AVRO-value"), schemaRegistryAddr, avroSchema).as("value"))
 .write
 .format("kafka")
 .option("kafka.bootstrap.servers", bootstrapServers)
 .option("kafka.security.protocol", "SSL")
 .option("kafka.ssl.keystore.location", kafkaKeystoreLocation)
 .option("kafka.ssl.keystore.password", keystorePassword)
 .option("kafka.ssl.truststore.location", kafkaTruststoreLocation)
 .option("topic",topic)
 .save()

我的问题是 - 如果 strDf.count 是 62M，我可以直接将其写入 Kafka 还是需要迭代 cmd# 3。

【问题讨论】：

我已经添加了上面的代码。我的问题是 - 如果 strDf.count 是 62M，我可以使用 Cmd3 直接将其写入 Kafka，还是需要将 Cmd# 3 迭代几次，比如 20 次......并在 1 次中写入 3M 块数据。
感谢您深入了解此 Mike。 Kafka 安装在本地，当我尝试从 Databricks Notebook 写入 Kafka 时，它超时。在 62M 中，写入了 490 万条记录。我写了一个批量 Kafka 生产者；不是火花海峡。流媒体。我使用批处理是因为我每天只给经纪人写一次信。流式处理会很小心，因为它会以微批处理的形式发送记录。就我而言，我将尝试使用循环进行多次迭代。至少我知道使用当前配置。一次我可以写约 450 万条记录。您认为这是正确的方法吗？

标签： apache-spark apache-kafka kafka-producer-api spark-structured-streaming spark-streaming-kafka

【解决方案1】：

使用适用于 Kafka 的 Spark 结构化流将数据存储到 Kafka 中没有限制。您将在下面看到，您的流式查询将创建一个（池）KafkaProducer，用于遍历您的 Dataframe 中的行。 Kafka可以处理这么多消息，而且没有限制。

值得注意的是，在这批消息实际写入代理之前，Kafka 会将一些消息缓冲成一批。这是通过KafkaProducer Configslinger.ms、batch.size 和max.request.size 的配置来控制的，因此根据您的整体设置调整这些设置可能会很有用。

这是`spark-kafka-sql`库的代码：

在内部，Spark 将在 InternalKafkaProducerPool.scala 中创建一个 KafkaProducers 池：

  private def createKafkaProducer(paramsSeq: Seq[(String, Object)]): Producer = {
    val kafkaProducer: Producer = new Producer(paramsSeq.toMap.asJava)
    if (log.isDebugEnabled()) {
      val redactedParamsSeq = KafkaRedactionUtil.redactParams(paramsSeq)
      logDebug(s"Created a new instance of KafkaProducer for $redactedParamsSeq.")
    }
    kafkaProducer
  }

然后您的查询将被转换为 RDD，并且对于每个分区，它都会遍历 KafkaWriter.scala 中的元素：

  queryExecution.toRdd.foreachPartition { iter =>
      val writeTask = new KafkaWriteTask(kafkaParameters, schema, topic)
      Utils.tryWithSafeFinally(block = writeTask.execute(iter))(
        finallyBlock = writeTask.close())
    }
  }

数据的实际生产将发生在KafkaWriteTask：

  def execute(iterator: Iterator[InternalRow]): Unit = {
    producer = Some(InternalKafkaProducerPool.acquire(producerConfiguration))
    val internalProducer = producer.get.producer
    while (iterator.hasNext && failedWrite == null) {
      val currentRow = iterator.next()
      sendRow(currentRow, internalProducer)
    }
  }

【讨论】：

这是spark-kafka-sql库的代码：

这是`spark-kafka-sql`库的代码：