الكاتب: وو تشى هوى، بكين شبكة للهاتف النقال ونظم البيانات مركز صيانة مهندس معماري، ماجستير في هندسة البرمجيات، جامعة بكين للبريد والاتصالات السلكية واللاسلكية، وكبار المهندسين. لسنوات عديدة تعمل في مجال تصميم بنية النظام، وتطوير البرمجيات، وتحليل البيانات، ومشغلي كبيرة الحفريات العمل.

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

شركات كبيرة لديها مجموعة كبيرة من البيانات ومجموعة متنوعة من الميزات، مثل جميع أنواع من مشروع القانون، مما يدل، عادة ما تكون كمية البيانات يوم واحد، ثم هناك مليار عشر. مع الطلب على تحليل الاحتياجات أعمال تجهيز البيانات في الوقت الحقيقي في تزايد مستمر، ولكن أيضا لدينا تجهيز البنية البيانات الكبيرة يجلب تحديا كبيرا، مع الإشارة إلى مثال على الشبكة ليتم التحقيق فيها، وتطبيقها على تجهيز البنية الفعلية، وغالبا بسبب تدفق البيانات في الوقت الحقيقي، مما أدى إلى تشغيل نظام غير مستقر وتشوهات مختلفة. من تطوير في الوقت الحقيقي معالجة البيانات الهندسة المعمارية كبيرة للخط، واستغرق ما يقرب من شهرين، بعد، وقد استقرت نظامنا الكثير من التحسين. في نهاية المطاف نحن نستخدم مجموعة من 10 خدمة، وتجهيز الوقت الحقيقي من البيانات على عشرة مليارات في اليوم الواحد، عدد الحقول لكل البيانات وهنا 100، أطول محتويات الحقل أكثر من 1000 بايت.

وهنا ما لدينا لتبادل البيانات في الوقت الحقيقي العملية كمية كبيرة من معالجة البيانات بشكل عام، ولخص صعودا وهبوطا.

• أهداف المشروع

على أساس عدد محدود من كتلة ملقم يستند على تحقيق أكثر من عشرة مليارات في اليوم، وأكثر من 20T من الجسم، ثم التجهيز في الوقت الحقيقي واحد. احتياجات محددة FTP هو جمع قائمة مفصلة عبارة عن مضاعفات خادم واحد، وسيتم تخزين في الوقت الحقيقي تجهيز البيانات في قاعدة البيانات للمستخدمين لتفاصيل Hbase على الفور من استعلام واحد، في حين أن Hdfs فاتورة تخزينها لتحليل حاليا.

• موارد الأجهزة

10 إلى x86 الخوادم ومستقل التكوين خرطوشة 16 وحدة المعالجة المركزية والذاكرة 128G، 2T بجد * 2 * 10،300G القرص الثابت (قرص النظام).

• بنية النظام

10 خوادم hadoop العنقودية، حيث الليمفاوية NameNode في وقت واحد كما جمع تركيب FTP والمسايل، حدد خمسة خوادم أخرى مثبتة كافكا، حارس الحديقة والعاصفة تحقيق البيانات في الوقت الحقيقي الكبيرة تيار تجهيز الهندسة المعمارية، للاستفادة من موارد الحوسبة العنقودية، التي خمسة خوادم غزل مجهزة كمية صغيرة من موارد الحوسبة، والمشاركة في الاحتياجات اليومية تحليل البيانات خارج الخط. خوادم الأربعة المتبقية قمنا بتركيب Hbase تلبية احتياجات تحت ثاني الاستعلام البيانات الكبيرة، طوبولوجيا النظام على النحو التالي:

طوبولوجيا نظام FIG.

• تنفيذ المشروع

1. استخدام التكنولوجيات ذات الصلة

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

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

كافكا هو إنتاجية عالية من توزيع نشر الإشتراك مراسلة النظام، على غرار كمية كبيرة من البيانات العازلة بركة، البيانات تدعم استهلاك متعدد المستخدمين. حارس الحديقة هي خدمة التنسيق مفتوح المصدر وزعت التطبيقات الموزعة، كتلة تخزين مشتركة بين الدولة مسؤولة عن بعض من مكونات المعلومات التزامن. عاصفة توزيعها في الوقت الحقيقي النظم، بما في ذلك نيمبوس والمشرف من عقدة عقدة رئيسية (من storm1.0، بزيادة قدرها نيمبوس عقدة النسخ الاحتياطي) الحوسبة، حارس الحديقة قيام الحاجة إلى الاعتماد على تزامن الدولة بين العقد. مكونات كتلة العاصفة:

• نيمبوس: العاصفة هي عقدة رئيسية من الكتلة، المسؤولة عن تخصيص الموارد وجدولة المهام.

• المشرف: العاصفة هي مجموعة عقدة الرقيق، ولاية لقبول نيمبوس المخصصة لتشغيل وإيقاف عملية جزء الخاصة بهم من إدارة عامل، وزعت العقد بالمعنى الحقيقي الحوسبة.

مكونات الكتلة الشكل الثاني العاصفة

تطبيق العاصفة ينطوي على مفهوم التنمية، تشارك جافا نموذج البرمجة في البرنامج:

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

• صنبور: توليد مكونات دفق البيانات المصدر في طوبولوجيا. صنبور عادة لقراءة البيانات من مصدر بيانات خارجي، ثم يتم تحويل البيانات المصدر إلى المناطق الداخلية من طوبولوجيا.

• بولت: تلقي عنصر معالجة البيانات طوبولوجيا ومن ثم تنفيذها. بولت يستطيع أن يؤدي وظيفة عملية تصفية، الاندماج، أي قاعدة بيانات عملية الكتابة.

• الصفوف (tuple): وحدة أساسية من تسليم الرسالة.

2. فتح مكونات مصدر التثبيت والتكوين

أ) تركيب المسايل والتكوين

Http://flume.apache.org/ تحميلها من حزمة التثبيت المسايل، الضغط، امباري إذا مدير Cloudera أو التثبيت، إلا من خلال صفحة إدارة تكوين التثبيت المقابلة. نحن فقط المسايل تثبيت مستقل، لم يتم تثبيت مجموعة المسايل، قائمة بذاتها كفاءة التصنيع المسايل هي عالية جدا، ونحن قادرون تماما على تلبية معالجة مطالب المليارات من قطعة من البيانات كل يوم، ولكن تجدر الإشارة إلى أنه من متانة المسايل هامة من سيئة للغاية، وغالبا عملية في، ولكن العملية لا يعالج بيانات حالة عالقة، أن نلاحظ ما يلي عند استخدام المسايل:

• مراقبة المسايل دليل لا يمكن أن يحتوي الدليل.

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

a1.sources.r1.ignorePattern = ^ (.) * \. تمة $

• المسايل ملف معاملة قد تحتوي على أحرف خاصة، مما يؤدي عملية المسايل عالقة. مجموعة قاء الطابع التعرف تجاهل تخطي:

a1.sources.r1.decodeErrorPolicy = تجاهل

• يحدث GC خلال خلال العملية من أخطاء الذاكرة المسايل التخلص منها في تكوين الذاكرة flume-env.sh (الافتراضي هو صغير)؛

تصدير JAVA_OPTS = "- Xms1024m -Xmx2048m -Dcom.sun.management.jmxremote"

• عندما يبدأ المسايل العودة -c لإعطاء التفاصيل الكاملة دليل ملف المسايل، وإلا التكوين flume-env.sh لا الحمل، وسوف تستخدم التكوين الافتراضي، على سبيل المثال، الأمر التالي لبدء دليل ملف كامل:

وكيل /hadoop/apache-flume-1.6.0-bin/bin/flume-ng /hadoop/apache-flume-1.6.0-bin/conf/ -c

• إذا طابور الذاكرة، يرجى ملاحظة أن عدد طابور تكوين الذاكرة الرسالة، يجب أن يكون حجم طابور أكبر مجموعة من أو يساوي transactionCapacity BatchSize و.

a1.channels.c1.transactionCapacity = 2000a1.sinks.k1.batchSize = 2000

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

• A طول المسايل قراءة سجل أكثر من 2048 حرفا، أي سيتم قطع، ويمكن زيادة 4096 بايت لحل العناصر التالية مرتبة في ملف التكوين:

producer.sources.s.deserializer.maxLineLength = 65535

• قضايا تحويل الأحرف غير طبيعية المسايل، java.nio.charset.MalformedInputException: مساهمة طول = 1، يمكنك إضافة عناصر التكوين التالية إلى عنوان في ملف التكوين:

a1.sources.r1.inputCharset = ISO8859-1

• واجه المسايل توقف مشوه، ذكرت الشاذ: java.nio.charset.MalformedInputException، يمكنك إضافة التكوين التالي في ملف التكوين، تجاهل البيانات خطأ (الافتراضي هو FAIL، ورمي خطأ استثناء، المسايل ستوقف) إلى حل.

producer.sources.s.decodeErrorPolicy = تجاهل

• افتراضيا، معالجة المجرى ملفات سيزيد .completed لاحقة، في كمية كبيرة من البيانات، وجمع تدعم قريبا القرص الصلب الكامل، يمكنك إضافة التكوين التالي في ملف التكوين، بحيث المسايل التعامل مع حذف تلقائيا ملفات البيانات حلها.

a1.sources.r1.deletePolicy = فوري

تكوين المسايل:

a1.sources = R1

التكوين Flume-env.sh:

# متغيرات بيئة يمكن تعيين هنا.

المسايل الأمر تشغيل:

وكيل /hadoop/apache-flume-1.6.0-bin/bin/flume-ng -c /hadoop/apache-flume-1.6.0-bin/conf/ -f /hadoop/apache-flume-1.6.0-bin /conf/viewdata.conf منتج -n -Dflume.root.logger = خطأ و

علما بأن مسار يجب أن تعطي كامل ملف التكوين المسايل، وإلا فإنه لا يمكن تحميل Flume-env.sh بداية المجرى تكوينها بشكل صحيح.

ب) تثبيت نظام كافكا والتكوين

كافكا تحميل حزمة التثبيت من :. بعد كافكا _ * TGZ، الضغط، ملف التكوين server.properties.

server.properties التكوين:

# معرف الجهاز في كافكا العنقودية

broker.id = 48

# ميناء الخدمة

ميناء = 9092

# اسم المضيف

host.name = storm01

# عدد المواضيع التعامل مع طلبات الشبكة

num.network.threads = 3

# عدد المواضيع القيام القرص I / O

num.io.threads = 8

# المخزن المؤقت الإرسال (SO_SNDBUF) المستخدمة من قبل الملقم مقبس

socket.send.buffer.bytes = 102400

# وتلقي العازلة (SO_RCVBUF) المستخدمة من قبل خادم المقبس

socket.receive.buffer.bytes = 102400

# الحد الأقصى لحجم طلب أن الملقم مأخذ ستقبل (حماية ضد OOM)

socket.request.max.bytes = 104857600

موقع تخزين البيانات #kafka (عندما كبيرة كمية الاحتياجات من البيانات ليتم تخزينها في الدليل أن يكون بالحجم الكامل)

log.dirs = / DATA1 / كافكا-السجلات

# عدد الافتراضي موضوع إنشاء قسم

num.partitions = 1

# ينصح هذه القيمة إلى زيادة للمنشآت مع تطبيق الاستعراض المفصل البيانات الموجودة في RAID مجموعة.

num.recovery.threads.per.data.dir = 1

الحدث #kafka فلاش فقط أن يكون من الصعب أن تتبع استهلاك المستهلك، وذلك لتكوين معلمات الوقت فلاش، لتجنب تحديث البيانات الوقت في حالة وجود كمية صغيرة من البيانات طويل جدا

log.flush.interval.messages = 10000

log.flush.interval.ms = 1000

تخزين # البيانات في الوقت كافكا، في ساعات، والعمل الإضافي سيتم تلقائيا البيانات كافكا حذفها

log.retention.hours = 48

# الحد الأقصى لحجم ملف شريحة السجل. وعند الوصول إلى هذا الحجم يتم إنشاء شريحة سجل جديد.

log.segment.bytes = 1073741824

# والفاصل الزمني الذي يتم فحص شرائح السجل لمعرفة ما اذا كان يمكن حذفها وفقا

لسياسات الاحتفاظ

log.retention.check.interval.ms = 300000

# إذا = log.cleaner.enable يتم تعيين نظافة الحقيقية سيتم تمكين ويمكن بعد ذلك السجلات الفردية يتم وضع علامة لضغط السجل.

log.cleaner.enable = كاذبة

# تكوين كتلة حارس الحديقة

zookeeper.connect = سيد: 2181، storm01: 2181، storm02: 2181، storm03: 2181، storm04: 2181

# مهلة في مللي ثانية لربط حارس الحديقة

zookeeper.connection.timeout.ms = 6000

# هل قادرة على حذف موضوع التكوين، كاذبة الافتراضي لا يمكن إزالة الموضوع

delete.topic.enable = صحيح

كافكا تشغيل الخدمة: JPS الأمر يمكن أن نرى كافكا اسم العملية، وقد بدأت وصف كافكا بنجاح.

nohup kafka-server-start.sh /home/hadoop/kafka_2.9.1-0.8.2.1/config/server.properties &

إنشاء موضوع: إنشاء كائن عامل تكرار 2، هناك قسم في الموضوع 24، وخلق وافر من التقسيم هو زيادة التوازي، نسخ بيانات كائن عامل التكرار الأمن.

kafka-topics.sh --create --zookeeper الماجستير: 2181، storm01: 2181، storm02: 2181، storm03: 2181، storm04: 2181 --replication عامل 2 --partitions 24 sighttp --topic

تخزين البيانات كافكا: في دليل تخزين البيانات كافكا، يمكنك ان ترى في كل - تتم تسمية مجلد، على سبيل المثال sighttp 19 ممثلة موضوع: sighttp والتقسيم: 19، كما هو مبين أدناه:

الشكل الثالث

موضوع التقسيم إلى الدليل، يمكنك ان ترى الكثير من الوثائق وتنتهي في .index .LOG. حيث البيانات ملف .LOG، والتي يتم تخزينها في كافكا تجمع المخزن المؤقت للبيانات، غير .index ملف فهرس، ملف البيانات وملف الفهرس في أزواج، اسم الملف هو عبارة عن سلسلة من الأرقام التي تحدد بداية البيانات المخزنة في ملف SEQ ID، على النحو التالي:

الشكل الرابع

كافكا استعلام حالة استهلاك البيانات: من يتم تسجيل كافكا بيانات استهلاك المستهلكين في حارس الحديقة الدولة، استخدم الأمر zkCli.sh إلى طريقة الاستعلام هو مبين أدناه موضوع الاستهلاك: sighttp والتقسيم: الدولة 0، ويقابل يشير التي تم تجهيزها ل49259227840 خط كما هو موضح أدناه:

الشكل V

الخبرة: الاستهلاك من قبل عدد من الصفوف لتخزين عدد الصفوف ليتم تحديد ما إذا كانت سرعة برنامج معالجة البيانات تحتاج سرعة توليد البيانات.

استهلاك كافكا استثناء النموذجية:

 خطأ: خطأ عند معالجة جلب طلب التقسيم تعويض 6535061966 من المستهلكين مع ارتباط معرف 0. السبب المحتمل: طلب تعويض 6535061966 ولكن ليس لدينا سوى شرائح السجل في نطاق 6580106664 إلى 6797636149. (kafka.server.ReplicaManager)

الشاذ: كافكا قد انتهت بالفعل بسبب العدد الرسالة يتم حذف 6535061966 الرسالة، النطاق الحالي ليست سوى كافكا 6580106664-6797636149 سجل، ولكن المستهلكين على صفقة صلاحية حذف الرسالة، أن يظهر هذه الرسالة استثناء (عادة منذ سرعة معالجة البيانات بطيئة، وسرعة توليد البيانات لا يمكن أن تلبي الاحتياجات، مما أدى إلى تراكم الرسائل، الروافد رسالة انتهاء الوقت المتراكمة كافكا تكوين لحذف كافكا).

ج) تركيب مجموعة العاصفة والتكوين

عاصفة في تحميل حزمة التثبيت، فمن المستحسن استخدام الإصدار 0.10.0 صدر العاصفة أعلاه، لأن أحدث الإصلاحات نسخة الكثير من الأخطاء، وخاصة قضية STORM-935 (بعد بدء طوبولوجيا يستخدم الكثير من موارد النظام ، مما أدى إلى عملية غير مستقرة طوبولوجيا).

storm.yaml ملف التكوين:

#zookeeper تكوين الكتلة Server

storm.zookeeper.servers: - "سيد" - "storm01" - "storm02" - "storm03" - "storm04"

عقدة رئيسية #storm

nimbus.host: "سيد"

خدمة #strom الصفحة إدارة الموانئ

ui.port: 8081

#storm من تكوين المنفذ العقدة الخدمة، الافتراضي 6700-6703 ما مجموعه أربعة منافذ، مما يعني أن كل الخادم يمكن أن توفر أربع فتحات عامل، حيث بزيادة قدرها 6704 و6705 الميناء، خادم واحد أي بزيادة اثنين من فتحات عامل ، عامل يعني زيادة في عدد مجموعات العاصفة يمكن أن توفر المزيد من الموارد الحاسوبية.

supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703 - 6704 - 6705

# الشرط موقع تخزين المعلومات، لتجنب استخدام / تمة

storm.local.dir: "/home/hadoop/apache-storm-0.10.0/workdir"

# الذاكرة عقدة ماستر

nimbus.childopts: "-Xmx3072m"

من ذاكرة عقدة #

supervisor.childopts: "-Xmx3072m"

ذاكرة #worker، ويمكن زيادة الذاكرة مشاكل GC الزائد يقلل

worker.childopts: "-Xmx3072m"

# الافتراضي هو 30، أي بزيادة قدرها مدة netty مهلة وغيرها من المعالم، والحد من مشاكل الاتصال بسبب Netty، مما تسبب عامل عدم الاستقرار

storm.messaging.netty.max_retries: 60

# إعدادات إضافة storm.messaging.netty.max_wait_ms، الافتراضي هو 1000

storm.messaging.netty.max_wait_ms: 2000

بدء تشغيل الخدمة:

• :( عقدة رئيسية لبدء عقدة وإدارة الخدمات الصفحة الرئيسية)

  nohup العاصفة نيمبوس و

  nohup العاصفة واجهة المستخدم و

• من العقدة: nohup المشرف العاصفة و

صفحة إدارة العاصفة:

متصفح العاصفة UI وأدخل عنوان حيث عدد منفذ خادم +8081، وفتح صفحة إدارة ستروم كما هو مبين أدناه:

الشكل السادس

من ملخص العنقودية يمكن أن يرى الشكل السادس في ما مجموعه 4 عقد نظام المجموعة المشرف العاصفة، لأن كل المشرف يوفر ستة فتحة (إذا لم يتم تكوين الملكية supervisor.slots.ports storm.yaml في ملف التكوين، شريطة كل 4 المشرف الافتراضي فتحة)، وبالتالي ما مجموعه 4 * 6 = 24 فتحة، 22 وقد استخدمت، هناك نوعان من الخمول. ملاحظة أنه بمجرد أن طوبولوجيا من كل إصدار، ستشغل طويلة فتحة، إذا لم يكن هناك ما يكفي من فتحة، الإصدار الأخير من طوبولوجيا ستشغل فتحة الخمول، وفتحة لايوجد احتلت الموارد الأخرى استباق، وإذا كان هناك أي فتحة، لن يتم نشر طوبولوجيا جديد، ثم تحتاج للاستفادة من ملقم كتلة العاصفة المحتملين، أو لإضافة موارد جديدة لزيادة فتحة ملف تكوين الملقم.

وكما يتبين من الشكل ملخص VI طوبولوجيا في الكتلة أصدرت سبعة طوبولوجيا، طوبولوجيا المحتلة من قبل كل الموارد عامل، وعدد من بدء موضوع المنفذ، على وجه التحديد كيف استهلاك الموارد بكثير المحدد في برنامج تطوير العاصفة طوبولوجيا.

د) كافكا + العاصفة + + Hdfs Hbase التنمية طوبولوجيا

نحن نستخدم الكسوف لإنشاء مشروع مخضرم، إضافة العاصفة وتابعيها في pom.xml ملف التكوين Hdfs، في هذه الحالة هو بيانات استهلاك عاصفة من كافكا، عملية ETL بعد تخزينها Hdfs وHbase، تحتاج إلى إضافة العاصفة كافكا، العاصفة Hdfs، العاصفة Hbase التي تعتمد على، الالتفات إلى تبعيات الإصدار إلى أن تكون متسقة مع الكتلة.

استخراج فئات عملية وراثة BaseRichBolt:

الطبقة العامة splitBolt يمتد BaseRichBolt {خاص ثابت النهائي سلسلة TAB = ""؛

جامع OutputCollector الخاص؛

الفراغ العام إعداد (خريطة التكوين، والسياق TopologyContext، OutputCollector جامع) {

this.collector = جامع؛}

الفراغ العام تنفيذ (المدخلات الصفوف (tuple)) {خط سلسلة = input.getString (0)؛ بعبارة سلسلة = line.split (TAB)؛

إذا (words.length > 74) {حساب سلسلة.

إذا (words.length > 0) = حساب الكلمات؛

آخر حساب = ""؛ سلسلة LocalIPv4.

إذا (words.length > 0) LocalIPv4 = الكلمات؛

آخر LocalIPv4 = ""؛ سلسلة RemoteIPv4.

إذا (words.length > 0) RemoteIPv4 = الكلمات؛

آخر RemoteIPv4 = ""؛ سلسلة السطر = + الحساب "|" + LocalIPv4 + "|" + RemoteIPv4، collector.emit (المدخلات، والقيم الجديدة (السطر الجديد))؛} collector.ack (المدخلات)؛}

declareOutputFields الفراغ الجمهور (OutputFieldsDeclarer المعلن عن) {declarer.declare (الحقول الجديدة ( "السطر"))؛}}

إرسال Hbase ضرورة تحقيق فئات HBaseMapper:

تنفذ الطبقة العامة myHbaseMapper HBaseMapper {الأعمدة ColumnList الجمهور (الصفوف (tuple) الصفوف (tuple)) {سلسلة خط = tuple.getString (0)؛ بعبارة سلسلة = line.split ( "\ |")؛ ColumnList العواميد = ColumnList جديدة.

// المعلمة تليها أسماء العائلة العمود، أسماء الأعمدة، والقيمة إذا (words.length > 0) cols.addColumn ( "المحتوى" .getBytes "LocalIPv4" .getBytes، words.getBytes)؛

إذا (words.length > 0) cols.addColumn ( "المحتوى" .getBytes "RemoteIPv4" .getBytes، words.getBytes)؛

العودة العواميد؛}

rowKey بايت العام (الصفوف (tuple) الصفوف (tuple)) {سلسلة خط = tuple.getString (0)؛ بعبارة سلسلة = line.split ( "\ |")؛ مفتاح سلسلة.

// rowkey ترتيب حساب المضادة سلسلة، لتسهيل أقسام جدول البيانات داخل hbase مفتاح التوازن = ب StringBuilder الجديد (كلمات) .reverse.toString.

العودة key.getBytes؛}}

وظائف رئيسية:

ساكنة العام باطلة الرئيسية (وسائط سلسلة) {ZKS سلسلة = "سيد: 2181، storm01: 2181، storm02: 2181"، // حارس الحديقة مجموعة سلسلة الموضوع = "topicname"؛ // كافكا اسم الموضوع سلسلة zkRoot = "/ العاصفة" ؛ // حارس الحديقة المعلومات الدليل الجذر المخزنة في معرف سلسلة الحالة = "kafkatopicname" معلومات ؛. دولة المخزنة في طوبولوجيا // حارس الحديقة دليل فرعي fileNameFormat fileNameFormat = جديد DefaultFileNameFormat .withPath ( "/ العاصفة / تمة /") withPrefix ( " تمة _ ") withExtension (." دات ")؛ شكل RecordFormat = جديد DelimitedRecordFormat .withFieldDelimiter (" | ") ؛. اسم الملف التسويقي // يكتب hdfs يبدأ مع" tmp_ '،' دات. '// نهاية كل 10 دقائق إعادة كتابة ملف جديد hdfs FileRotationPolicy rotationPolicy = TimedRotationPolicy الجديدة (10.0f، TimeUnit.MINUTES)؛ brokerHosts brokerHosts = ZkHosts الجديدة (ZKS)؛

أمثلة من الإجراء أعلاه لتحقيق العاصفة وقراءة كافكا الكتابة Hdfs Hbase، يجوز أن تستند استخراج على فئات مختلفة من متطلبات الخدمة، لتحقيق مختلف منطق رمز جافا. مرفوع حزمة جرة جمعت إلى الكتلة، باستخدام العاصفة تقدم سطر الأوامر البنية:

عاصفة جرة ./kafkastream.jar sighdfs.sighttphdfs stormmaster

ملخص

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

وأخيرا، فإننا نشارك اثنين من المشاكل العملية التي تواجهها:

• حارس الحديقة تكوين طوبولوجيا عملية غير المستقرة الناجمة عن العاصفة

مجموعة العاصفة بسبب حارس الحديقة الحاجة الكتلة لمزامنة الدولة، لذلك كل عملية العامل الخادم هو عاصفة سوف تبقي الاتصال عقدة حارس الحديقة، حارس الحديقة عقدة العدد الافتراضي من الاتصالات هو 60، عندما العاصفة حساب عدد كبير من طبولوجيا، حارس الحديقة حاجة لتعديل maxClientCnxns التكوين = 1000، أي بزيادة اتصالات حارس الحديقة.

• Hdfs عقدة القرص I / O طوبولوجيا تشغيل عدم الاستقرار عالية الناجمة عن العاصفة

منذ يتم احتساب العاصفة في الوقت الحقيقي، فإن كل من الازدحام طوبولوجيا العاصفة الجانب يسبب عدم استقرار في تطوير عقدة نواجه Hdfs القرص I / O عالية، مما أدى إلى عاصفة الكتابة Hdfs العمل الإضافي، وتؤدي في النهاية إلى عامل قتل مشرف، مما أدى إلى مشكلة الطوبوغرافية من عدم الاستقرار. والسبب هو أن العقد على إحدى Hdfs، غزل لحد من عملية قيد التنفيذ، -x 110 مع أمر للتحقق من iostat، الوقت / O غزل القرص وسيطة I هو احتلت 100، في حين أن البيانات غزل لوحة المتوسطة هي Hdfs لوحة، مما أدى إلى طلب الكتابة تفشل في الاستجابة، مما يؤدي في النهاية إلى Hdfs الكتابة عاصفة من الوقت بدل الضائع عامل، مما تسبب في طوبولوجيا عملية غير مستقر. عندما اقترح تكوين لوحة هنا غزل وسيطة، لا تستخدم القرص جذر نظام التشغيل، لا تستخدم قرص بيانات Hdfs، يمكنك تجنب فعال لمشكلة Hdfs العاصفة الكتابة المهلة.