Как я могу отправлять большие сообщения с Kafka (более 15 МБ)?

Вам нужно настроить три (или четыре) свойства:

• Сторона потребителя: fetch.message.max.bytes - определяет максимальный размер сообщения, которое может получить потребитель.
• Сторона брокера: replica.fetch.max.bytes - это позволит репликам в брокерах отправлять сообщения в кластере и обеспечивать правильную репликацию сообщений. Если это слишком мало, то сообщение никогда не будет реплицировано, и, следовательно, потребитель никогда не увидит сообщение, потому что сообщение никогда не будет зафиксировано (полностью реплицировано).
• Сторона брокера: message.max.bytes - это наибольший размер сообщения, которое может быть получено брокером от производителя.
• Сторона брокера (для каждой темы): max.message.bytes - это наибольший размер сообщения, которое брокер разрешает добавлять в тему. Этот размер подтвержден предварительным сжатием. (По умолчанию message.max.bytes брокера.)

Я узнал о номере 2 на собственном горьком опыте - вы не получаете НИКАКИХ исключений, сообщений или предупреждений от Kafka, поэтому обязательно учитывайте это, когда отправляете большие сообщения.

Незначительные изменения необходимы для Kafka 0.10 и нового потребителя по сравнению с ответ смеющегося_мана:

• Брокер: Без изменений, вам все равно нужно увеличить свойства message.max.bytes и replica.fetch.max.bytes. message.max.bytes должен быть равен или меньше (*) replica.fetch.max.bytes.
• Производитель: увеличьте max.request.size, чтобы отправить более крупное сообщение.
• Потребитель: увеличьте max.partition.fetch.bytes, чтобы получать сообщения большего размера.

(*) Прочтите комментарии, чтобы узнать больше о _7 _ ‹= _ 8_

Ответ от @laughing_man довольно точен. Но все же я хотел дать рекомендацию, которую я узнал от эксперта Kafka Стефана Маарека. Мы активно применяем это решение в наших живых системах.

Kafka не предназначен для обработки больших сообщений.

Ваш API должен использовать облачное хранилище (Ex AWS S3) и просто отправить в Kafka или любой брокер сообщений ссылку на S3. Вы должны найти место для хранения ваших данных, может быть, это сетевой диск, может быть, что угодно, но это не должен быть брокер сообщений.

Теперь, если вы не хотите использовать вышеуказанное решение

Максимальный размер сообщения - 1 МБ (настройка в ваших брокерах называется message.max.bytes) Apache Kafka. Если вам это действительно нужно, вы можете увеличить этот размер и обязательно увеличить сетевые буферы для ваших производителей и потребителей.

И если вы действительно заботитесь о разделении вашего сообщения, убедитесь, что каждое разделение сообщения имеет один и тот же ключ, чтобы оно было перенесено в один и тот же раздел, а содержимое вашего сообщения должно сообщать «идентификатор части», чтобы ваш потребитель мог полностью восстановить сообщение. .

Вы также можете изучить сжатие, если ваше сообщение основано на тексте (сжатие gzip, snappy, lz4), что может уменьшить размер данных, но не волшебным образом.

Опять же, вам нужно использовать внешнюю систему для хранения этих данных и просто отправить внешнюю ссылку на Kafka. Это очень распространенная архитектура, и вам следует придерживаться ее, и она широко распространена.

Имейте в виду, что Kafka лучше всего работает, только если сообщения огромны по объему, но не по размеру.

Источник: https://www.quora.com/How-do-I-send-Large-messages-80-MB-in-Kafka

Идея состоит в том, чтобы иметь одинаковый размер сообщения, отправляемого от Kafka Producer в Kafka Broker, а затем полученного Kafka Consumer, т.е.

Производитель Kafka -> Брокер Kafka -> Потребитель Kafka

Предположим, что если требуется отправить 15 МБ сообщения, то нужно синхронизировать Producer, Broker и Consumer. .

Kafka Producer отправляет 15 МБ -> Kafka Broker разрешает / сохраняет 15 МБ -> Kafka Потребитель получает 15 МБ.

Таким образом, настройка должна быть:

а) о Брокере:

message.max.bytes=15728640 
replica.fetch.max.bytes=15728640

б) на Потребителе:

fetch.message.max.bytes=15728640

Вам необходимо переопределить следующие свойства:

Конфигурации брокера ($ KAFKA_HOME / config / server.properties)

• replica.fetch.max.bytes
• message.max.bytes

Конфигурации потребителей ($ KAFKA_HOME / config / consumer.properties)
Этот шаг у меня не сработал. Я добавляю его в потребительское приложение, и оно работает нормально

• fetch.message.max.bytes

Перезагрузите сервер.

дополнительную информацию см. в этой документации: http://kafka.apache.org/08/configuration.html

Важно помнить, что атрибут message.max.bytes должен быть синхронизирован со свойством потребителя fetch.message.max.bytes. размер выборки должен быть не меньше максимального размера сообщения, в противном случае может возникнуть ситуация, когда производители могут отправлять сообщения большего размера, чем потребитель может потреблять / извлекать. Возможно, стоит взглянуть на это.
Какую версию Kafka вы используете? Также предоставьте более подробную информацию о трассировке, которую вы получаете. есть ли что-то вроде ... payload size of xxxx larger than 1000000 в журнале?

Для людей, использующих landoop kafka: вы можете передавать значения конфигурации в переменных среды, например:

docker run -d --rm -p 2181:2181 -p 3030:3030 -p 8081-8083:8081-8083  -p 9581-9585:9581-9585 -p 9092:9092
 -e KAFKA_TOPIC_MAX_MESSAGE_BYTES=15728640 -e KAFKA_REPLICA_FETCH_MAX_BYTES=15728640  landoop/fast-data-dev:latest `

И если вы используете rdkafka, то передайте message.max.bytes в конфигурации производителя, например:

  const producer = new Kafka.Producer({
        'metadata.broker.list': 'localhost:9092',
        'message.max.bytes': '15728640',
        'dr_cb': true
    });

Аналогичным образом для потребителя

  const kafkaConf = {
   "group.id": "librd-test",
   "fetch.message.max.bytes":"15728640",
   ... .. }                                                                                                                                                                                                                                                      

Я думаю, что большинство ответов здесь устарели или не совсем полны.

Чтобы обратиться к ответу Саши Веттер (с обновлением для Kafka 0.10), я Я хочу предоставить дополнительную информацию и ссылки на официальную документацию.

Конфигурация производителя:

• max.request.size (Link) необходимо увеличить для файлов размером более 1 МБ, в противном случае они отклоняются

Конфигурация брокера / темы:

• message.max.bytes (Link) может быть установлен, если вы хотите увеличить сообщение размер на уровне брокера. Но из документации: Это можно установить для каждой темы с конфигурацией max.message.bytes на уровне темы.
• max.message.bytes (Link) можно увеличить, если только одна тема должна поддерживать принимать лагерные файлы. Конфигурацию брокера изменять нельзя.

Я всегда предпочитаю конфигурацию с ограничением темы, потому что я могу настроить тему самостоятельно как клиент для кластера Kafka (например, с помощью клиент администратора). Я не могу иметь никакого влияния на саму конфигурацию брокера.

В ответах выше при необходимости упоминаются еще несколько конфигураций:

• replica.fetch.max.bytes (ссылка) (конфигурация брокера)

Из документации: Это не абсолютный максимум, если первый пакет записей в первом непустом разделе выборки больше этого значения, пакет записей все равно будет возвращен, чтобы гарантировать, что прогресс может быть достигнут.

• max.partition.fetch.bytes (ссылка) (потребительская конфигурация)

Из документации: Записи загружаются заказчиком партиями. Если первый пакет записей в первом непустом разделе выборки превышает этот предел, пакет все равно будет возвращен, чтобы гарантировать, что потребитель может продолжить работу.

• fetch.max.bytes (Link) (Consumer config; не упоминается выше, но из той же категории )

Из документации: Записи выбираются в пакетах потребителем, и если первый пакет записей в первом непустом разделе выборки превышает это значение, пакет записей все равно будет возвращен, чтобы гарантировать, что потребитель может добиться прогресса.

Заключение. Не нужно изменять конфигурации получения сообщений для обработки сообщений, больше, чем значения по умолчанию для этой конфигурации (если это было протестировано в небольшой установке). Вероятно, потребитель всегда может получить пакеты размером 1. Однако необходимо установить две конфигурации из первого блока, как упоминалось в ответах ранее.

Это пояснение не должно ничего рассказывать о производительности и не должно быть рекомендацией устанавливать или не устанавливать эту конфигурацию. Лучшие значения необходимо оценивать индивидуально в зависимости от конкретной запланированной пропускной способности и структуры данных.

Вот как я успешно отправил данные размером до 100 МБ с помощью kafka-python==2.0.2:

Маклер:

consumer = KafkaConsumer(
    ...
    max_partition_fetch_bytes=max_bytes,
    fetch_max_bytes=max_bytes,         
)

Производитель (см. Окончательное решение в конце):

producer = KafkaProducer(
    ...
    max_request_size=KafkaSettings.MAX_BYTES,
)

Потом:

producer.send(topic, value=data).get()

После отправки таких данных появилось следующее исключение:

MessageSizeTooLargeError: The message is n bytes when serialized which is larger than the total memory buffer you have configured with the buffer_memory configuration.

Наконец, я увеличил buffer_memory (по умолчанию 32 МБ), чтобы получать сообщение на другом конце.

producer = KafkaProducer(
    ...
    max_request_size=KafkaSettings.MAX_BYTES,
    buffer_memory=KafkaSettings.MAX_BYTES * 3,
)