如何在使用 Spark 结构化流时更新 Kafka 消费者 max.request.size 配置答案

【问题标题】：How to update Kafka consumer max.request.size config while using Spark structured stream如何在使用 Spark 结构化流时更新 Kafka 消费者 max.request.size 配置
【发布时间】：2018-01-16 21:16:21
【问题描述】：

用于 Kafka 的 Spark readStream 失败并出现以下错误：

org.apache.kafka.common.errors.RecordTooLargeException（消息序列化时为 1166569 字节，大于最大值您使用 max.request.size 配置的请求大小配置。）

我们如何提升max.request.size？

代码：

val ctxdb = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "ip:port")
  .option("subscribe","topic")
  .option("startingOffsets", "earliest")
  .option(" failOnDataLoss", "false")
  .option("max.request.size", "15728640")

我们尝试更新option("max.partition.fetch.bytes", "15728640")，但没有成功。

【问题讨论】：

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

【解决方案1】：

您需要在写入器流设置中添加kafka 前缀：

.option("kafka.max.request.size", "15728640")

【讨论】：

感谢 Yuval，用“kafka”。前缀配置选项正在传递给 kafka 消费者。
但是，目前仍然收到相同的错误消息大小，并且无法识别上述选项“WARN consumer.ConsumerConfig: The configuration 'max.request.size' is provided but is not a known config 。” kakfa 版本：0.10.2.1
@user2286963 因为那是一个 producer 配置。你在使用 Spark 中内置的 kafka 生产者吗？
@user2286963 那么你正在寻找fetch.message.max.bytes。此外，您需要确保在代理上配置了具有正确请求大小的主题，我假设您已经设置了该大小，因为您的生产者可以生成这些大文件。
Yuval，请忽略上述cmets。我从 1 个主题中阅读，并写给许多主题。我用 option("kafka.max.request.size", "15728640") 配置了 writer 并且它可以工作。