كيف يمكنني إرسال رسائل كبيرة مع كافكا (أكثر من 15 ميغابايت)?

Question

أرسل رسائل سلسلة إلى كافكا الخامس.0.8 مع أبي منتج جافا.إذا كان حجم الرسالة حوالي 15 ميغابايت أحصل على MessageSizeTooLargeException.لقد حاولت تعيين message.max.bytesإلى 40 ميغابايت ، ولكن ما زلت أحصل على الاستثناء.عملت الرسائل الصغيرة دون مشاكل.

(يظهر الاستثناء في المنتج ، ليس لدي مستهلك في هذا التطبيق.)

ماذا يمكنني أن أفعل للتخلص من هذا الاستثناء?

بلدي مثال منتج التكوين

private ProducerConfig kafkaConfig() {
    Properties props = new Properties();
    props.put("metadata.broker.list", BROKERS);
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("request.required.acks", "1");
    props.put("message.max.bytes", "" + 1024 * 1024 * 40);
    return new ProducerConfig(props);
}

خطأ-سجل:

4709 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with correlation id 214 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
4869 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with    correlation id 217 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5035 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with   correlation id 220 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5198 [main] WARN  kafka.producer.async.DefaultEventHandler  - Produce request with correlation id 223 failed due to [datasift,0]: kafka.common.MessageSizeTooLargeException
5305 [main] ERROR kafka.producer.async.DefaultEventHandler  - Failed to send requests for topics datasift with correlation ids in [213,224]

kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(Unknown Source)
at kafka.producer.Producer.send(Unknown Source)
at kafka.javaapi.producer.Producer.send(Unknown Source)

Answer 1

تحتاج إلى ضبط ثلاث (أو أربع) خصائص:

• جانب المستهلك:fetch.message.max.bytes - سيحدد هذا أكبر حجم للرسالة التي يمكن للمستهلك جلبها.
• جانب الوسيط: replica.fetch.max.bytes - سيسمح هذا للنسخ المتماثلة في الوسطاء بإرسال رسائل داخل المجموعة والتأكد من نسخ الرسائل بشكل صحيح.إذا كان هذا صغيرا جدا ، فلن يتم تكرار الرسالة أبدا ، وبالتالي ، لن يرى المستهلك الرسالة أبدا لأن الرسالة لن يتم الالتزام بها أبدا (يتم نسخها بالكامل).
• جانب الوسيط: message.max.bytes - هذا هو أكبر حجم للرسالة التي يمكن أن يتلقاها الوسيط من المنتج.
• جانب الوسيط (لكل موضوع): max.message.bytes - هذا هو أكبر حجم للرسالة التي سيسمح الوسيط بإلحاقها بالموضوع.يتم التحقق من صحة هذا الحجم قبل الضغط.(التخلف عن الوسيط message.max.bytes.)

لقد اكتشفت الطريقة الصعبة حول الرقم 2-لا تحصل على أي استثناءات أو رسائل أو تحذيرات من كافكا ، لذا تأكد من مراعاة ذلك عند إرسال رسائل كبيرة.

Answer 2

تغييرات طفيفة مطلوبة ل كافكا 0.10 و ال مستهلك جديد مقارنة ب إجابة الرجل الضاحك:

• الوسيط:لا توجد تغييرات، لا تزال بحاجة إلى زيادة الخصائص message.max.bytes و replica.fetch.max.bytes. message.max.bytes يجب أن تكون مساوية أو أصغر (*) من replica.fetch.max.bytes.
• منتج:يزيد max.request.size لإرسال الرسالة الأكبر.
• مستهلك:يزيد max.partition.fetch.bytes لتلقي رسائل أكبر.

(*) إقرأ التعليقات لتعرف المزيد عنها message.max.bytes<=replica.fetch.max.bytes

Answer 3

تحتاج إلى تجاوز الخصائص التالية:

تكوين الوسيط ($ kafka_home / config / server.properties)

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

التكوينات الاستهلاكية ($ kafka_home / config / consicer.properties)
<م> هذه الخطوة لم تنجح بالنسبة لي.أضفها إلى تطبيق المستهلك وكان يعمل بشكل جيد

• fetch.message.max.bytes

أعد تشغيل الخادم.

إلقاء نظرة على هذه الوثائق لمزيد من المعلومات: http://kafka.apache.org/08/configuration.html

Answer 4

تتمثل الفكرة في الحصول على حجم متساوٍ من الرسالة التي يتم إرسالها من منتج Kafka إلى Kafka Broker ثم يتم استلامها بواسطة Kafka Consumer، أي.

منتج كافكا --> كافكا وسيط --> كافكا المستهلك

لنفترض أنه إذا كان الشرط هو إرسال 15 ميغابايت من الرسالة، فإن منتج, ، ال وسيط و ال مستهلك, يجب أن تكون الثلاثة متزامنة.

منتج كافكا يرسل 15 ميغابايت --> وسيط كافكا يسمح/يخزن 15 ميجابايت --> كافكا المستهلك يتلقى 15 ميغابايت

لذلك يجب أن يكون الإعداد:

أ) على الوسيط:

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

ب) على المستهلك:

fetch.message.max.bytes=15728640

Answer 5

شيء رئيسي واحد أن نتذكر أن message.max.bytes يجب أن تكون السمة في المزامنة مع المستهلك fetch.message.max.bytes الملكية.يجب أن يكون حجم الجلب على الأقل بحجم الحد الأقصى لحجم الرسالة وإلا فقد يكون هناك موقف يمكن للمنتجين فيه إرسال رسائل أكبر مما يمكن للمستهلك استهلاكه/جلبه.قد يستحق الأمر إلقاء نظرة عليه.
أي إصدار من كافكا الذي تستخدمه?أيضا تقديم بعض مزيد من التفاصيل تتبع التي تحصل عليها.هل هناك شيء مثل ... payload size of xxxx larger than 1000000 الخروج في السجل?

Answer 6

الجواب من @ يضحك رجل دقيق جدا.لكن مع ذلك ، أردت أن أقدم توصية تعلمتها من خبير كافكا ستيفان معارك من قرة.

ليس المقصود كافكا للتعامل مع الرسائل الكبيرة.

يجب أن تستخدم واجهة برمجة التطبيقات الخاصة بك التخزين السحابي (إكس أوس س 3) ، ودفع فقط إلى كافكا أو أي وسيط رسالة مرجع س 3.يجب أن تجد في مكان ما لاستمرار البيانات الخاصة بك ، وربما انها محرك أقراص الشبكة ، وربما انها أيا كان ، ولكن لا ينبغي أن يكون وسيط الرسالة.

الآن ، إذا كنت لا تريد الذهاب مع الحل أعلاه

الحد الأقصى لحجم الرسالة هو 1 ميغابايت (يسمى الإعداد في الوسطاء message.max.bytes) أباتشي كافكا.إذا كنت حقا في حاجة إليها بشدة ، يمكنك زيادة هذا الحجم والتأكد من زيادة المخازن المؤقتة للشبكة للمنتجين والمستهلكين.

وإذا كنت تهتم حقا بتقسيم رسالتك ، فتأكد من أن كل تقسيم للرسالة يحتوي على نفس المفتاح بالضبط بحيث يتم دفعه إلى نفس القسم ، ويجب أن يبلغ محتوى رسالتك عن "معرف جزء" حتى يتمكن المستهلك من إعادة بناء الرسالة بالكامل.

يمكنك أيضا استكشاف ضغط ، إذا كانت رسالتك يستند إلى نص (غزيب ، لاذع ، لز 4 ضغط) والتي قد تقلل من حجم البيانات ، ولكن ليس بطريقة سحرية.

مرة أخرى ، لديك لاستخدام نظام خارجي لتخزين تلك البيانات ومجرد دفع إشارة خارجية إلى كافكا.هذا هو بنية شائعة جدا ، واحد يجب أن تذهب مع ومقبولة على نطاق واسع.

ضع ذلك في اعتبارك أن كافكا يعمل بشكل أفضل فقط إذا كانت الرسائل ضخمة الحجم ولكن ليس في الحجم.

المصدر: https://www.quora.com/How-do-I-send-Large-messages-80-MB-in-Kafka