使用spark structured streaming for kafka将数据存储到kafka中没有限制。您将在下面看到，您的流式查询将创建一个（池） KafkaProducer 用于遍历 Dataframe . Kafka可以处理如此多的信息，而且没有限制。
值得注意的是，kafka将在将这批消息实际写入代理之前将一些消息缓冲到一个批中。这是通过kafkaproducer配置的配置来引导的 linger.ms , batch.size 以及 max.request.size ，因此根据您的总体设置调整这些设置可能很有用。

以下是spark kafka sql库的代码：

在内部，spark将在internalkafkaproducerpool.scala中创建一个kafkaproducer池：

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)
    }
  }