HDInsight में कमजोर संरचित डेटा को संसाधित करने के लिए MapReduce

इस उदाहरण में, हम Microsoft के क्लाउड-आधारित Hadoop कार्यान्वयन में एक विशिष्ट MapReduce कार्य बनाने और निष्पादित करने पर ध्यान देंगे, जिसे HDInsight कहा जाता है।
पिछले उदाहरण में, हमने 3-नोड हडोप क्लस्टर बनाया और थोड़ा संरचित प्रारूप का सार लॉग लोड किया, जिसे अब संसाधित किया जाना है। पत्रिका आम तौर पर एक बड़ी (हमारे विशिष्ट उदाहरण में, छोटी है, लेकिन यह विचार के मौलिक प्रदर्शन को प्रभावित नहीं करती है) पाठ फ़ाइल जिसमें संकेत TRACE, DEBUG, INFO, WARN, ERROR, FATAL के साथ लाइनें हैं। हमारा प्राथमिक कार्य प्रत्येक विशेषता के साथ पंक्तियों की संख्या की गणना करना होगा, अर्थात WARN की स्थिति कितनी बार हुई, कितने ERROR इत्यादि हुए?

SQL के संदर्भ में, आपको COUT () ... GROUP BY को विशेषता फ़ील्ड द्वारा करने की आवश्यकता है। यह स्पष्ट है कि जैसे कोई फ़ील्ड नहीं है, चूंकि फ़ाइल एक प्लेट नहीं है, लेकिन समस्या के एक पाठ विवरण के साथ लाइनों का एक सेट है, जिसमें लक्षण के नाम के साथ एक विकल्प है। आपको सभी लाइनों पर जाने की जरूरत है, विशेषता का एक विकल्प चुनें और जोड़ें। सीधे शब्दों में कहें, से

2012-02-03 18:35:34 SampleClass6 [INFO] everything normal for id 577725851 2012-02-03 18:35:34 SampleClass4 [FATAL] system problem at id 1991281254 2012-02-03 18:35:34 SampleClass3 [DEBUG] detail for id 1304807656 2012-02-03 18:35:34 SampleClass3 [WARN] missing id 423340895 2012-02-03 18:35:34 SampleClass5 [TRACE] verbose detail for id 2082654978 2012-02-03 18:35:34 SampleClass0 [ERROR] incorrect id 1886438513 ... 

लिपि १

आपको कुछ पाने की जरूरत है

 [TRACE] 10 [DEBUG] 20 [INFO] 30 [WARN] 555 [ERROR] 777 [FATAL] 1 

लिपि २

MapReduce मॉडल के पीछे का विचार बहुत सरल है। एक वितरित प्रणाली की उपस्थिति में, जो एक Hadoop क्लस्टर है, सामान्य कार्य को समानांतर उप-कार्यों में विभाजित (मैप) किया जाता है। जैसा कि पिछले उदाहरण में बताया गया है, Hadoop फ़ाइल सिस्टम को सहेजते समय उपयोगकर्ता को उपयोगकर्ता के लिए नोड्स द्वारा पारदर्शी रूप से टुकड़ों में विभाजित किया जाता है। सैद्धांतिक रूप से, इन नोड्स को भौगोलिक रूप से वितरित किया जा सकता है, अर्थात। विभिन्न भौगोलिक स्थानों में हो। डेटा केंद्रों (या बस अलग नोड्स के बीच) के बीच डेटा स्थानांतरित करने से जुड़ी लागतों को कम करने के लिए, Hadoop डेटा की क्षेत्रीय निकटता को ध्यान में रखता है - प्रत्येक उप-कार्य अपने स्वयं के डेटा के साथ काम करता है। हमारे मामले में, क्लस्टर में केवल 3 नोड हैं, लक्जरी नहीं। असाइनमेंट को उन नोड्स पर निष्पादित किया जाएगा जहां डेटा टुकड़े झूठ बोलते हैं। उपकुंजी के परिणामों को फिर उपयोगकर्ता के लिए लौटाए गए एकल परिणाम में कटौती कार्यों द्वारा एकत्र किया जाता है। दूसरे शब्दों में, प्रत्येक नोड अपना निजी उप-परिणाम देगा, उदाहरण के लिए, पहला -

 [TRACE] 1 [DEBUG] 2 [INFO] 3 ... 

लिपि ३

दूसरा -

 [TRACE] 9 [DEBUG] 5 [INFO] 7 ... 

लिपि ४

जिनमें से अंत में समग्र परिणाम संकलित किया जाएगा। स्क्रिप्ट 2. यह समानांतर प्रसंस्करण का एक सामान्य विचार है, जिसे पारंपरिक रिलेशनल डेटाबेस सर्वर (उदाहरण के लिए, ओरेकल आरएसी, माइक्रोसॉफ्ट एसक्यूएल सर्वर समानांतर डेटावेयर, आदि) और क्लाउड रिलेशनल सेवाओं में लागू किया गया था। डेटा प्रोसेसिंग (उदा।, Windows Azure SQL डेटाबेस में एक फ़ेडरेटेड डेटाबेस, जिसे पहले SQL Azure में शार्किंग के रूप में जाना जाता है)। लेकिन इस मामले में, हम एक रिलेशनल के साथ काम नहीं कर रहे हैं, लेकिन एक खराब संरचित इनपुट डेटा प्रारूप है, इसलिए, SQL स्क्रिप्ट के बजाय, हमें ऐसे कार्यों को लिखना होगा जो मानचित्र / भूमिका को कम करते हैं। MapReduce का विचार विभिन्न भाषाओं में लागू किया गया है। उदाहरण के लिए, मुफ्त खुला स्रोत अपाचे हडोप परियोजना इन उद्देश्यों के लिए जावा का उपयोग करती है। चूंकि Microsoft HDInsight अपाचे Hadoop के साथ संगत है, इसलिए हम जावा भाषा और org.apache.hadoop.mapreduce पैकेज का भी उपयोग करेंगे।

प्रारंभ में, मैप क्लास, जिसे मैपर से लिया गया है, लागू किया गया है। मैपर वर्ग कुंजी / मान जोड़े के एक प्रारंभिक सेट को एक मध्यवर्ती में परिवर्तित करता है। हमारे मामले में, इनपुट मान पाठ लॉग फ़ाइल की रेखाएं हैं - मानचित्र विधि के पाठ प्रकार का मान पैरामीटर। विधि के अंदर, प्रत्येक मान में, हम वर्ग कोष्ठक की तलाश करते हैं, बीच में जो है उसे बाहर निकालें, संकेतों के निरंतर सेट के साथ तुलना करें, जिसे हम शुरुआत में पैटर्न चर में डालते हैं और, अगर यह मेल खाता है (यदि (matcher.matches ())), तो हम आउटपुट की-जोड़ी बनाते हैं। मूल्य। कुंजी साइन TRACE / DEBUG / ... (पाठ चर logLevel) का विकल्प है, और मान 1। मान IntWritable प्रकार के संचायक चर में निहित है, जिसे हमने एक में कंस्ट्रक्टर में आरंभीकृत किया है। IntWriteable एक Java int प्रकार के आसपास एक आवरण है, जो Writable इंटरफ़ेस लागू करता है। Hadoop अपने क्रमांकन प्रारूप का उपयोग करता है। हम प्रत्येक विशेषता की घटनाओं की संख्या की गणना करने के लिए इन लोगों को कटौती समारोह में जोड़ देंगे। मध्यवर्ती (आउटपुट) मान को प्रत्येक आउटपुट कुंजी के लिए Hadoop वातावरण द्वारा समूहीकृत किया जाता है। मैपिंग स्टेज पर, आप डेटा को कम करने के लिए setCombinerClass के साथ पूर्व-एकत्र कर सकते हैं। Reducer को प्रेषित किया गया। इस उदाहरण में, इस सुविधा का उपयोग नहीं किया गया है। रिपोर्टर वर्ग (मानचित्र विधि का अंतिम पैरामीटर) का उद्देश्य निष्पादन की स्थिति और प्रगति, अद्यतन काउंटरों आदि को प्रदर्शित करना है। हमारे सरल उदाहरण में, इसका उपयोग भी नहीं किया जाता है।
Reducer से निकाली गई Reduce class, उलटा समस्या हल करती है। यह मैपिंग के मध्यवर्ती परिणाम एकत्र करता है और उन्हें एकत्र करता है, इस मामले में कुख्यात COUNT () मानों का प्रदर्शन करता है, क्योंकि मैपिंग के दौरान चाबियों के आधार पर ग्रुप बाय (छँटाई सहित) किया गया। कम करने के लिए इनपुट प्रकार (पाठ, आंतरिक) मानचित्र से आउटपुट प्रकारों से मेल खाना चाहिए। रिड्यूस चरण में परिणामों के विलय के दौरान, हडोपॉप पर्यावरण एक द्वितीयक प्रकार का प्रदर्शन करता है, क्योंकि विभिन्न मैपर से प्राप्त परिणामों में एक ही कुंजी हो सकती है। इस प्रकार, कटौती विधि के लिए इनपुट कुंजी तार का एक सेट है - इसके अनुरूप मूल्यों का एक संग्रह। उदाहरण के लिए, लाइनों में से एक TRACE (कुंजी) और कई इकाइयों का एक संग्रह होगा क्योंकि इस विशेषता के होने की संख्या मैपर के एक या दूसरे उदाहरण द्वारा निर्धारित की गई थी। यह हमारे लिए संग्रह पर जाने और चर गणना में योग रखने के लिए रहता है। OutputCollector में हम पारंपरिक की-वैल्यू पेयर लिखते हैं, यहाँ केवल वैल्यू कुंजी द्वारा एकत्रीकरण का परिणाम होगा।
मुख्य () विधि का उपयोग बनाए गए मानचित्र के आधार पर एक Hadoop नौकरी बनाने और कक्षाओं को कम करने और इसे निष्पादित करने के लिए किया जाता है। JobConf ऑब्जेक्ट एक कार्य विनिर्देश बनाता है। कोड एक JAR फ़ाइल को लिखा जाता है जिसे Hadoop क्लस्टर में वितरित करेगा। फ़ाइल नाम को स्पष्ट रूप से निर्दिष्ट करने के बजाय, आप एन्कोडिंग क्लास को एग्जीक्यूटेबल कोड (MapReduceTest) से जॉबकोनफ कंस्ट्रक्टर को पास कर सकते हैं, जिसके द्वारा Hadoop को संबंधित JAR फाइल मिल जाएगी। SetOutputKeyClass () और setOutputValueClass () तरीके मैप और रिड्यूस फ़ंक्शन के लिए आउटपुट प्रकार सेट करते हैं। एक नियम के रूप में, वे संयोग करते हैं, अर्थात्। नक्शा Reduce के समान ही उत्पादन करता है। यदि वे अलग-अलग हैं, तो MapMap फ़ंक्शन का आउटपुट प्रकार setMapOutputKeyClass () और setMapOutputValueClass () विधियों का उपयोग करके निर्दिष्ट किया जा सकता है। कौन सा क्लास मैप करेगा, और कौन सा कम करें, जैसा कि आप अनुमान लगा सकते हैं, setMapperClass () और setReducerClass () विधियों का उपयोग करके सेट किया गया है। यह इनपुट / आउटपुट प्रारूप को निर्धारित करने के लिए रहता है। यह setInputFormat () और setOutputFormat () विधियों द्वारा किया जाता है। इस मामले में, यह नहीं किया जा सका, क्योंकि पाठ प्रारूप डिफ़ॉल्ट रूप से स्वीकार किया जाता है। अंत में, आपको स्टैटिक विधियों FileInputFormat.setInputPaths () और FileOutputFormat.setOutputPath () का उपयोग करके स्रोत डेटा और परिणामों के साथ फ़ाइलों को पथ निर्दिष्ट करने की आवश्यकता है। हम कमांड लाइन तर्कों के माध्यम से फ़ाइल नाम पारित करेंगे। जैसा कि विधि का नाम दिखाता है, कई इनपुट फाइलें हो सकती हैं। शायद एक निर्देशिका, फिर इसमें निहित सभी फाइलें ली जाएंगी। आप एक फ़ाइल नाम टेम्पलेट निर्दिष्ट कर सकते हैं। एक निर्देशिका उस स्थान के रूप में निर्दिष्ट की जाती है जहाँ परिणाम फ़ाइलों को जोड़ा जाएगा। यह मौजूद नहीं होना चाहिए, अन्यथा निष्पादन के दौरान एक त्रुटि होगी। एक प्रकार का सुरक्षा उपाय ताकि एक कार्य दूसरे के परिणाम को न भटकाए। आप हैडूप fs -rmr कमांड का उपयोग करके एक निर्देशिका को हटा सकते हैं।
इसे एक साथ रखने पर, हमें निम्नलिखित कोड मिलते हैं:

 //   import java.io.IOException; import java.util.Iterator; import java.util.regex.Matcher; import java.util.regex.Pattern; //,   Hadoop import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapred.FileInputFormat; import org.apache.hadoop.mapred.FileOutputFormat; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapred.TextInputFormat; import org.apache.hadoop.mapred.TextOutputFormat; public class MapReduceTest { /* *  */ public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> { private static final Pattern pattern = Pattern.compile("(TRACE)|(DEBUG)|(INFO)|(WARN)|(ERROR)|(FATAL)"); //   private static final IntWritable accumulator = new IntWritable(1); //   - ,    private Text logLevel = new Text(); public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> collector, Reporter reporter) throws IOException { //    '['  ']' final String[] tokens = value.toString().split("[ \\[\\]]"); if(tokens != null) { //  logLevel for(final String token : tokens) { final Matcher matcher = pattern.matcher(token); if(matcher.matches()) //  { logLevel.set(token); collector.collect(logLevel, accumulator); //  - } } } } } /* *  */ public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> { public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> collector, Reporter reporter) throws IOException { int count = 0; //  count    while(values.hasNext()) { count += values.next().get(); } System.out.println(key + "\t" + count); collector.collect(key, new IntWritable(count)); } } /* *   */ public static void main(String[] args) throws Exception { //       ,  Map/Reduce final JobConf conf = new JobConf(MapReduceTest.class); conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(IntWritable.class); conf.setMapperClass(Map.class); conf.setReducerClass(Reduce.class); conf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); //-       FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); //  JobClient.runJob(conf); } } 

स्क्रिप्ट 5

दूरस्थ डेस्कटॉप के माध्यम से Hadoup क्लस्टर पर चलते हैं, जैसा कि पिछले लेख में दिखाया गया है, और इस कोड को MapReduceTest.java फ़ाइल में सहेजें, उसी d: \ Temp में कहें। HDInsight के Java सपोर्ट लाइब्रेरी C: \ apps \ java \ bin में स्थित हैं। इस बारे में Hadoop को जानकारी नहीं है। यह Hadoop कमांड प्रॉम्प्ट विंडो में जाने के लिए समझ में आता है (

 D:\Windows\system32\cmd.exe /k pushd "c:\apps\dist\hadoop-1.1.0-SNAPSHOT" && "c:\apps\dist\hadoop-1.1.0-SNAPSHOT\bin\hadoop.cmd" 

, सुविधा के लिए, HDInsight डेस्कटॉप पर एक शॉर्टकट है) और इस पथ को% पथ% परिवेश चर पर लिखें:

 set PATH=%PATH%;C:\apps\java\bin 

लिपि ६

D: \ Temp निर्देशिका पर जाएं और Java फ़ाइल को बायटेकोड क्लास फ़ाइलों में संकलित करें। जब से मैंने यूनिकोड एन्कोडिंग में MapReduceTest.java को सहेजा है, तब -encoding स्विच की आवश्यकता थी।

 javac -encoding UNICODE -classpath C:\apps\dist\hadoop-1.1.0-SNAPSHOT\hadoop-core-*.jar d:\Temp\MapReduceTest.java 

लिपि 7

D: \ Temp में, MapReduceTest.class फ़ाइल और MapReduceTest $ Map.class और MapReduceTest $ Reduce.class वर्गों के अनुरूप फ़ाइलें बनाई जाती हैं। विधानसभा का निर्माण करें:

 jar -cvf MapReduceTest.jar *.class 

लिपि 8


चित्र 1

वर्तमान पथ d: \ Temp पर, Java संग्रह MapReduceTest.jar का गठन किया गया था।

 hadoop jar MapReduceTest.jar MapReduceTest Sample1/input/Sample.log Sample1/output 

लिपि ९


Fig.2

यहाँ नमूना 1 / इनपुट / सेम्पलॉग लॉग फ़ाइल है जिसे स्थानीय d: \ Temp निर्देशिका से HDFS / नमूना 1 / इनपुट निर्देशिका में डाउनलोड किया जाना है - पिछले लेख के चित्र 5 देखें। पिछली बार, मैं इस तथ्य पर ध्यान केंद्रित करना भूल गया था कि डाउनलोड करने से पहले एचडीएफएस इनपुट डायरेक्टरी (हडूप fs -mkdir Sample1 / input /) को स्पष्ट रूप से बनाना आवश्यक है और उसके बाद ही फाइल (हडूप fs -put d) \ Temp \ Sample.log डालें नमूना 1 / इनपुट /)। यदि आप किसी निर्देशिका को बनाए बिना किसी फ़ाइल को डाउनलोड करने का प्रयास करते हैं, तो इसे बनाया जाता है, लेकिन फ़ाइल इसमें लोड नहीं होती है, जैसा कि आप हडप fs -ls Sample1 / input / देख सकते हैं।
इस बीच, कार्य सफलतापूर्वक पूरा हुआ। आउटपुट निर्देशिका HDFS नमूना 1 / आउटपुट में, एक फ़ाइल को परिणाम में लॉग में प्रत्येक विशेषता की घटनाओं की संख्या वाले परिणामों के साथ उत्पन्न किया गया था:

 hadoop fs -cat Sample1/output/part-00000 

लिपि १०


3 चित्र