Как обновить конфигурацию max.request.size потребителя Kafka при использовании структурированного потока Spark

Spark readStream для Kafka выдает следующие ошибки:

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-kafka apache-spark spark-structured-streaming kafka-consumer-api
person user2286963    schedule 09.08.2017    source источник

Ответы (1)


arrow_upward
3
arrow_downward

Вам нужно добавить префикс kafka в настройку потока записи:

.option("kafka.max.request.size", "15728640")
person Yuval Itzchakov    schedule 09.08.2017
comment
Спасибо Юваль, с кафкой. префикс, параметры конфигурации передаются потребителю kafka. - person user2286963; 09.08.2017
comment
Но в настоящее время по-прежнему возникает та же ошибка по размеру сообщения, и указанный выше параметр не распознается WARN consumer.ConsumerConfig: конфигурация max.request.size была предоставлена, но не является известной конфигурацией. какфа версия: 0.10.2.1 - person user2286963; 09.08.2017
comment
@ user2286963 Потому что это конфигурация производителя. Вы используете встроенный производитель кафки в Spark? - person Yuval Itzchakov; 09.08.2017
comment
Наши производители внешние, и они хорошо продвигаются. Наши потребители терпят поражение. server.properties: message.max.bytes=10000000 replica.fetch.max.bytes=15728640 max.request.size=15728641 Какую опцию мне нужно использовать на стороне потребителя, чтобы он читал большие сообщения? - person user2286963; 09.08.2017
comment
@ user2286963 Значит, вы ищете fetch.message.max.bytes. Кроме того, вам необходимо убедиться, что ваша тема настроена на брокерах с правильным размером запроса, который, я предполагаю, вы уже установили, поскольку ваш производитель может создавать эти большие файлы. - person Yuval Itzchakov; 09.08.2017
comment
Yuval, пожалуйста, не обращайте внимания на приведенные выше комментарии. Я читаю из одной темы и пишу многим. Я настроил писатель с опцией (kafka.max.request.size, 15728640), и он работает. - person user2286963; 09.08.2017