في هذه المدونة سوف نستكشف إمكانيات وإمكانيات أحد أهم مكونات تقنية Hadoop ، أي MapReduce.
اليوم ، تتبنى الشركات إطار عمل Hadoop كخيارها الأول لتخزين البيانات نظرًا لقدرتها على التعامل مع البيانات الكبيرة بفعالية. لكننا نعلم أيضًا أن البيانات متعددة الاستخدامات وموجودة في هياكل وأشكال مختلفة. للتحكم في مثل هذا التنوع الهائل من البيانات وأشكالها المختلفة ، يجب أن تكون هناك آلية لاستيعاب جميع الأنواع ، ومع ذلك إنتاج نتيجة فعالة ومتسقة.
أقوى مكون في إطار عمل Hadoop هو MapReduce الذي يمكن أن يوفر التحكم في البيانات وهيكلها بشكل أفضل من نظرائه الآخرين. على الرغم من أنه يتطلب زيادة في منحنى التعلم وتعقيد البرمجة ، إذا كان بإمكانك التعامل مع هذه التعقيدات ، يمكنك بالتأكيد التعامل مع أي نوع من البيانات باستخدام Hadoop.
يقسم إطار عمل MapReduce جميع مهام المعالجة الخاصة به إلى مرحلتين أساسيتين: Map و Reduce.
يتطلب إعداد بياناتك الأولية لهذه المراحل فهم بعض الفئات والواجهات الأساسية. الطبقة الممتازة لإعادة المعالجة هذه هي نمط الإدخال.
ال نمط الإدخال class هي إحدى الفئات الأساسية في Hadoop MapReduce API. هذه الفئة مسؤولة عن تحديد شيئين رئيسيين:
- تقسيم البيانات
- قارئ التسجيلات
تقسيم البيانات هو مفهوم أساسي في إطار عمل Hadoop MapReduce الذي يحدد كلاً من حجم مهام الخريطة الفردية وخادم التنفيذ المحتمل. ال قارئ السجل مسؤول عن قراءة السجلات الفعلية من ملف الإدخال وإرسالها (كأزواج مفتاح / قيمة) إلى مصمم الخرائط.
يتم تحديد عدد مصممي الخرائط بناءً على عدد الانقسامات. إن مهمة InputFormat هي إنشاء التقسيمات. في معظم الأوقات ، يكون حجم الانقسام مكافئًا لحجم الكتلة ، ولكن لا يتم دائمًا إنشاء التقسيمات بناءً على حجم كتلة HDFS. يعتمد الأمر كليًا على كيفية تجاوز طريقة getSplits () لتنسيق InputFormat.
نطاق دقة المشغل c ++
هناك فرق جوهري بين MR Split و HDFS block. الكتلة عبارة عن جزء فعلي من البيانات بينما يكون الانقسام مجرد جزء منطقي يقرأه مصمم الخرائط. لا يحتوي الانقسام على بيانات الإدخال ، فهو يحتوي فقط على مرجع أو عنوان للبيانات. يحتوي الانقسام بشكل أساسي على شيئين: الطول بالبايت ومجموعة من مواقع التخزين ، وهي مجرد سلاسل.
لفهم هذا بشكل أفضل ، دعنا نأخذ مثالاً واحدًا: معالجة البيانات المخزنة في MySQL باستخدام MR. نظرًا لعدم وجود مفهوم للكتل في هذه الحالة ، فإن النظرية: 'يتم إنشاء الانقسامات دائمًا بناءً على كتلة HDFS' ،فشل. أحد الاحتمالات هو إنشاء تقسيمات بناءً على نطاقات من الصفوف في جدول MySQL (وهذا ما يفعله DBInputFormat ، وهو تنسيق إدخال لقراءة البيانات من قواعد البيانات العلائقية). قد يكون لدينا عدد k من الانقسامات التي تتكون من n من الصفوف.
فقط بالنسبة إلى تنسيقات InputFormat التي تستند إلى FileInputFormat (تنسيق InputFormat لمعالجة البيانات المخزنة في الملفات) يتم إنشاء التقسيمات بناءً على الحجم الإجمالي لملفات الإدخال بالبايت. ومع ذلك ، يتم التعامل مع حجم كتل نظام الملفات لملفات الإدخال كحد أعلى لتقسيمات الإدخال. إذا كان لديك ملف أصغر من حجم كتلة HDFS ، فستحصل على مصمم خرائط واحد فقط لهذا الملف. إذا كنت تريد أن يكون لديك سلوك مختلف ، يمكنك استخدام mapred.min.split.size. لكن الأمر يعتمد مرة أخرى فقط على getSplits () لتنسيق InputFormat الخاص بك.
لدينا العديد من تنسيقات الإدخال الموجودة مسبقًا المتاحة ضمن الحزمة org.apache.hadoop.mapreduce.lib.input.
CombineFileRecordReaderWrapper.html
CombineSequenceFileInputFormat.html
SequenceFileAsBinaryInputFormat.html
SequenceFileAsTextInputFormat.html
SequenceFileAsTextRecordReader.html
الافتراضي هو TextInputFormat.
وبالمثل ، لدينا العديد من تنسيقات الإخراج التي تقرأ البيانات من المخفضات وتخزنها في HDFS:
PartialFileOutputCommitter.html
SequenceFileAsBinaryOutputFormat.html
كيفية المرور بالرجوع في جافا
الافتراضي هو TextOutputFormat.
بحلول الوقت الذي تنتهي فيه من قراءة هذه المدونة ، تكون قد تعلمت:
- كيفية كتابة برنامج تقليل الخريطة
- حول الأنواع المختلفة من تنسيقات InputFormats المتوفرة في Mapreduce
- ما هي الحاجة إلى InputFormats
- كيفية كتابة تنسيقات InputFormats المخصصة
- كيفية نقل البيانات من قواعد بيانات SQL إلى HDFS
- كيفية نقل البيانات من قواعد بيانات SQL (هنا MySQL) إلى قواعد بيانات NoSQL (هنا Hbase)
- كيفية نقل البيانات من قواعد بيانات SQL إلى جدول آخر في قواعد بيانات SQL (ربما لا يكون هذا مهمًا جدًا إذا قمنا بذلك في نفس قاعدة بيانات SQL. ومع ذلك ، فلا حرج في الحصول على معرفة بنفس الشيء. أنت لا تعرف أبدًا كيف يمكن استخدامها)
المتطلبات المسبقة:
- Hadoop مثبت مسبقًا
- SQL مثبتة مسبقًا
- Hbase مثبتة مسبقا
- فهم جافا الأساسي
- MapReduce المعرفة
- إطار المعرفة الأساسية Hadoop
دعنا نفهم بيان المشكلة الذي سنحله هنا:
لدينا جدول موظف في MySQL DB في قاعدة البيانات العلائقية Edureka. الآن وفقًا لمتطلبات العمل ، يتعين علينا تحويل جميع البيانات المتاحة في DB العلائقية إلى نظام ملفات Hadoop ، أي HDFS و NoSQL DB المعروف باسم Hbase.
لدينا العديد من الخيارات للقيام بهذه المهمة:
- سكوب
- فلوم
- MapReduce
الآن ، لا تريد تثبيت أي أداة أخرى وتكوينها لهذه العملية. لديك خيار واحد فقط وهو MapReduce إطار معالجة Hadoop. يمنحك إطار عمل MapReduce التحكم الكامل في البيانات أثناء النقل. يمكنك معالجة الأعمدة ووضعها مباشرة في أي من الموقعين المستهدفين.
ملحوظة:
- نحتاج إلى تنزيل موصل MySQL ووضعه في مسار فئة Hadoop لجلب الجداول من جدول MySQL. للقيام بذلك ، قم بتنزيل الرابط com.mysql.jdbc_5.1.5.jar واحتفظ به ضمن دليل Hadoop_home / share / Hadoop / MaPreduce / lib.
تنزيلات cp / com.mysql.jdbc_5.1.5.jar $ HADOOP_HOME / share / hadoop / mapreduce / lib /
- أيضًا ، ضع جميع عبوات Hbase ضمن Hadoop classpath من أجل جعل برنامج MR الخاص بك يصل إلى Hbase. للقيام بذلك ، قم بتنفيذ الأمر التالي :
cp $ HBASE_HOME / lib / * $ HADOOP_HOME / مشاركة / hadoop / mapreduce / lib /
إصدارات البرامج التي استخدمتها في تنفيذ هذه المهمة هي:
- Hadooop-2.3.0
- HBase 0.98.9-Hadoop2
- كسوف القمر
من أجل تجنب البرنامج في أي مشكلة تتعلق بالتوافق ، أوصي القراء بتشغيل الأمر مع بيئة مماثلة.
اختيار فرز البرنامج في جافا
مخصص DBInputWritable:
package com.inputFormat.copy import java.io.DataInput import java.io.DataOutput import java.io.IOException import java.sql.ResultSet import java.sql.PreparedStatement import java.sql.SQLException import org.apache.hadoop.io .Writable import org.apache.hadoop.mapreduce.lib.db.DBWritable فئة عامة DBInputWritable implements Writable ، DBWritable {private int id الخاص String name ، dept public void readFields (DataInput in) يطرح IOException {} public void readFields (ResultSet rs) throws SQLException // يمثل كائن Resultset البيانات التي تم إرجاعها من جملة SQL {id = rs.getInt (1) name = rs.getString (2) dept = rs.getString (3)} كتابة عامة باطلة (DataOutput out) تلقي IOException { } تؤدي الكتابة العامة الباطلة (PreparedStatement ps) إلى SQLException {ps.setInt (1، id) ps.setString (2، name) ps.setString (3، dept)} public int getId () {return id} public String getName () {return name} public String getDept () {return dept}}مخصص DBOutputWritable:
package com.inputFormat.copy import java.io.DataInput import java.io.DataOutput import java.io.IOException import java.sql.ResultSet import java.sql.PreparedStatement import java.sql.SQLException import org.apache.hadoop.io .Writable import org.apache.hadoop.mapreduce.lib.db.DBWritable class DBOutputWritable implements Writable، DBWritable {private String name private int id private String dept public DBOutputWritable (String name، int id، String dept) {this.name = name this.id = id this.dept = dept} يطرح readFields العامة الباطلة (DataInput in) IOException {} حقول قراءة فارغة عامة (ResultSet rs) تطرح SQLException {} كتابة عامة باطلة (DataOutput out) IOException {} كتابة عامة باطلة (PreparedStatement ps) يطرح SQLException {ps.setString (1، name) ps.setInt (2، id) ps.setString (3، dept)}}جدول الإدخال:
إنشاء قاعدة بيانات edureka
إنشاء جدول emp (empid int not null ، اسم varchar (30) ، dept varchar (20) ، المفتاح الأساسي (empid))
أدخل في قيم emp (1 ، 'أبهاي' ، 'تطوير') ، (2 ، 'براندش' ، 'اختبار')
حدد * من إمبراطورية
الحالة 1: التحويل من MySQL إلى HDFS
الحزمة com.inputFormat.copy import java.net.URI import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.fs.Path import org.apache.hadoop.mapreduce استيراد الوظيفة org.apache.hadoop.mapreduce.lib.db.DBConfiguration import org.apache.hadoop.mapreduce.lib.db.DBInputFormat import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat import org.apache.hadoop .io.Text import org.apache.hadoop.io.IntWritable public class MainDbtohdfs {public static void main (String [] args) يطرح استثناء {Configuration conf = new Configuration () DBConfiguration.configureDB (conf، 'com.mysql.jdbc .Driver '، // driver class' jdbc: mysql: // localhost: 3306 / edureka '، // db url' root '، // user name' root ') // password Job job = new Job (conf) job .setJarByClass (MainDbtohdfs.class) job.setMapperClass (Map.class) job.setMapOutputKeyClass (Text.class) job.setMapOutputValueClass (IntWritable.class) job.setInputFormatClass (DBInputFormat.cormat FileOutput) مسار جديد (args [0])) DBInputFormat.setInput (وظيفة ، DBInputWritable.class ، 'emp' ، // اسم جدول الإدخال فارغ ، فارغ ، سلسلة جديدة [] {'empid'، 'name'، 'dept'} / / أعمدة الجدول) المسار p = مسار جديد (args [0]) FileSystem fs = FileSystem.get (عنوان URI جديد (args [0]) ، conf) fs.delete (p) System.exit (job.waitForCompletion (صحيح)؟ 0: 1)}}يتيح لنا هذا الجزء من التعليمات البرمجية إعداد أو تكوين صيغة الإدخال للوصول إلى مصدر قاعدة بيانات SQL لدينا. تتضمن المعلمة فئة برنامج التشغيل وعنوان URL يحتوي على عنوان قاعدة بيانات SQL واسم المستخدم وكلمة المرور.
DBConfiguration.configureDB (conf، 'com.mysql.jdbc.Driver'، // فئة السائق 'jdbc: mysql: // localhost: 3306 / edureka'، // db url 'root'، // اسم المستخدم 'root') //كلمه السر
يسمح لنا هذا الجزء من التعليمات البرمجية بتمرير تفاصيل الجداول في قاعدة البيانات وتعيينها في كائن الوظيفة. تتضمن المعلمات بالطبع مثيل الوظيفة ، والفئة القابلة للكتابة المخصصة التي يجب أن تنفذ واجهة DBWritable ، واسم جدول المصدر ، والشرط إذا كان أي شيء آخر فارغًا ، وأي معلمات فرز أخرى خالية ، وقائمة أعمدة الجدول على التوالي.
DBInputFormat.setInput (وظيفة ، DBInputWritable.class ، 'emp' ، // اسم جدول الإدخال فارغ ، فارغ ، سلسلة جديدة [] {'empid' ، 'name' ، 'dept'} // أعمدة الجدول)مخطط
الحزمة com.inputFormat.copy import java.io.IOException import org.apache.hadoop.mapreduce.Mapper import org.apache.hadoop.io.LongWritable import org.apache.hadoop.io.Text import org.apache.hadoop.io .IntWritable public class Map تعمل على توسيع Mapper {خريطة باطلة محمية (مفتاح LongWritable ، قيمة DBInputWritable ، سياق ctx) {try {String name = value.getName () IntWritable id = new IntWritable (value.getId ()) String dept = value.getDept ()ctx.write (نص جديد (الاسم + '' + معرف + '' + قسم) ، معرف)
} catch (IOException e) {e.printStackTrace ()} catch (InterruptException e) {e.printStackTrace ()}}}المخفض: مخفض الهوية المستخدم
أمر للتشغيل:
hadoop jar dbhdfs.jar com.inputFormat.copy.MainDbtohdfs / dbtohdfs
الإخراج: تم تحويل MySQL Table إلى HDFS
hadoop dfs -ls / dbtohdfs / *
الحالة 2: التحويل من جدول واحد في MySQL إلى جدول آخر في MySQL
إنشاء جدول الإخراج في MySQL
إنشاء جدول الموظف 1 (الاسم varchar (20) ، معرف int ، قسم varchar (20))
package com.inputFormat.copy import org.apache.hadoop.conf.Configuration import org.apache.hadoop.mapreduce.Job import org.apache.hadoop.mapreduce.lib.db.DBConfiguration import org.apache.hadoop.mapreduce.lib .db.DBInputFormat import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat import org.apache.hadoop.io.Text import org.apache.hadoop.io.IntWritable import org.apache.hadoop.io.NullWritable class public class Mainonetable_to_other_table {public static void main (String [] args) يطرح استثناء {Configuration conf = new Configuration () DBConfiguration.configureDB (conf، 'com.mysql.jdbc.Driver'، // driver class 'jdbc: mysql: // localhost : 3306 / edureka '، // db url' root '، // user name' root ') // password Job job = new Job (conf) job.setJarByClass (Mainonetable_to_other_table.class) job.setMapperClass (Map.class) job .setReducerClass (Reduce.class) job.setMapOutputKeyClass (Text.class) job.setMapOutputValueClass (IntWritable.class) job.setOutputKeyClass (DBOutputWritable.class) job.setOutputValueClass (Nul) lWritable.class) job.setInputFormatClass (DBInputFormat.class) job.setOutputFormatClass (DBOutputFormat.class) DBInputFormat.setInput (job، DBInputWritable.class، 'emp'، // input table name null، null، new String [] {'empid '،' name '،' dept '} // أعمدة الجدول) DBOutputFormat.setOutput (job،' worker1 '، // output table name new String [] {' name '،' id '،' dept '} // table أعمدة) System.exit (job.waitForCompletion (صحيح)؟ 0: 1)}}يتيح لنا هذا الجزء من التعليمات البرمجية تكوين اسم جدول الإخراج في SQL DB. المعلمات هي مثيل الوظيفة واسم جدول الإخراج وأسماء عمود الإخراج على التوالي.
DBOutputFormat.setOutput (job، 'worker1'، // output table name new String [] {'name'، 'id'، 'dept'} // أعمدة الجدول)مصمم الخرائط: نفس الحالة 1
المخفض:
الحزمة com.inputFormat.copy import java.io.IOException import org.apache.hadoop.mapreduce.Reducer import org.apache.hadoop.io.Text import org.apache.hadoop.io.IntWritable import org.apache.hadoop.io .NullWritable public class Reduce extends Reducer {protected void reduction (Text key، Iterable value، Context ctx) {int sum = 0 String line [] = key.toString (). split ('') try {ctx.write (new DBOutputWritable (سطر [0] .toString () ، Integer.parseInt (سطر [1] .toString ()) ، سطر [2] .toString ()) ، NullWritable.get ())} catch (IOException e) {e.printStackTrace ()} catch (InterruptException e) {e.printStackTrace ()}}}أمر للتشغيل:
hadoop jar dbhdfs.jar com.inputFormat.copy.Mainonetable_to_other_table
الإخراج: البيانات المنقولة من جدول EMP في MySQL إلى موظف جدول آخر في MySQL
الحالة 3: التحويل من جدول في MySQL إلى جدول NoSQL (Hbase)
إنشاء جدول Hbase لاستيعاب الإخراج من جدول SQL:
إنشاء 'موظف' ، 'معلومات رسمية'
فئة السائق:
الحزمة Dbtohbase import org.apache.hadoop.conf.Configuration import org.apache.hadoop.mapreduce.Job import org.apache.hadoop.mapreduce.lib.db.DBConfiguration import org.apache.hadoop.mapreduce.lib.db.DBInputFormat import org.apache.hadoop.hbase.mapreduce.TableOutputFormat import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.hbase.client.HTable import org.apache.hadoop.hbase.client.HTableInterface import org.apache .hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil import org.apache.hadoop.io.Text فئة عامة MainDbToHbase {public static void main (String [] args) يطرح استثناء {Configuration conf = HBaseConfiguration.create () HTableInterface mytable = new HTable (conf، 'emp') DBConfiguration.configureDB (conf، 'com.mysql.jdbc.Driver'، // فئة السائق 'jdbc: mysql: // localhost: 3306 / edureka' ، // db url 'root'، // user name 'root') // password Job job = new Job (conf، 'dbtohbase') job.setJarByClass (MainDbToHbase.class) job.s etMapperClass (Map.class) job.setMapOutputKeyClass (ImmutableBytesWritable.class) job.setMapOutputValueClass (Text.class) TableMapReduceUtil. فئة) DBInputFormat.setInput (وظيفة ، DBInputWritable.class ، 'emp' ، // اسم جدول الإدخال فارغ ، فارغ ، سلسلة جديدة [] {'empid' ، 'name' ، 'dept'} // أعمدة الجدول) (job.waitForCompletion (صحيح)؟ 0: 1)}}يتيح لك هذا الجزء من الكود تكوين فئة مفتاح الإخراج التي تكون في حالة hbase غير قابلة للتغيير
job.setMapOutputKeyClass (ImmutableBytesWritable.class) job.setMapOutputValueClass (Text.class)
هنا نقوم بتمرير اسم جدول hbase والمخفض للعمل على الطاولة.
TableMapReduceUtil.initTableReducerJob ('موظف' ، فئة تقليل ، وظيفة)مصمم الخرائط:
الحزمة Dbtohbase import java.io.IOException import org.apache.hadoop.mapreduce.Mapper import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.io .LongWritable import org.apache.hadoop.io.Text import org.apache.hadoop.io.IntWritable public class Map تمتد Mapper {private IntWritable one = new IntWritable (1) free void map (LongWritable id، DBInputWritable، Context Context) {try {String line = value.getName () String cd = value.getId () + '' String dept = value.getDept () Context.write (new ImmutableBytesWritable (Bytes.toBytes (cd)) ، new Text (line + ' '+ dept))} catch (IOException e) {e.printStackTrace ()} catch (InterruptException e) {e.printStackTrace ()}}}في هذا الجزء من الكود ، نأخذ قيمًا من حاصل فئة DBinputwritable ثم نمررها
غير قابلة للتغيير: قابلة للكتابة بحيث تصل إلى المخفض في شكل تربيعي بايت يفهمه Hbase.
String line = value.getName () String cd = value.getId () + '' String dept = value.getDept () Context.write (new ImmutableBytesWritable (Bytes.toBytes (cd)) ، نص جديد (سطر + '' + قسم ))
المخفض:
الحزمة Dbtohbase import java.io.IOException import org.apache.hadoop.hbase.client.Put import org.apache.hadoop.hbase.io.ImmutableBytesWritable import org.apache.hadoop.hbase.mapreduce.TableReducer import org.apache.hadoop .hbase.util.Bytes import org.apache.hadoop.io.Text public class Reduce extends TableReducer {public void reduction (ImmutableBytesWritable key، Iterable value، Context Context) بإلقاء IOException و InterruptException {سلسلة [] سبب = خالية // قيم حلقة لـ (Text val: القيم) {reason = val.toString (). split ('')} // ضع في HBase Put put = new Put (key.get ()) put.add (Bytes.toBytes ('official_info') ) ، Bytes.toBytes ('name') ، Bytes.toBytes (سبب [0])) put.add (Bytes.toBytes ('official_info') ، Bytes.toBytes ('القسم') ، Bytes.toBytes (سبب [1] ])) Context.write (key، put)}}يتيح لنا هذا الجزء من الكود تحديد الصف والعمود بدقة حيث سنخزن القيم من المخفض. نحن هنا نقوم بتخزين كل إمبيد في صف منفصل لأننا جعلنا مفتاح الصف الذي سيكون فريدًا. في كل صف نقوم بتخزين المعلومات الرسمية للموظفين تحت عمود العائلة 'official_info' تحت العمودين 'الاسم' و 'القسم' على التوالي.
ضع put = new Put (key.get ()) put.add (Bytes.toBytes ('official_info') ، Bytes.toBytes ('name') ، Bytes.toBytes (سبب [0])) put.add (Bytes. toBytes ('official_info') ، Bytes.toBytes ('القسم') ، Bytes.toBytes (السبب [1])) Context.write (مفتاح ، وضع)البيانات المحولة في Hbase:
فحص الموظف
كما نرى ، تمكنا من إكمال مهمة ترحيل بيانات أعمالنا من قاعدة بيانات SQL علائقية إلى قاعدة بيانات NoSQL DB بنجاح.
في المدونة التالية سنتعلم كيفية كتابة وتنفيذ الأكواد لتنسيقات الإدخال والمخرجات الأخرى.
استمر في نشر تعليقاتك أو أسئلتك أو أي ملاحظات. أحب أن أسمع منك.
لديك سؤال لنا؟ يرجى ذكر ذلك في قسم التعليقات وسنعاود الاتصال بك.
المنشورات ذات الصلة: