MapReduce उदाहरण

Question

मै मैड्रिड के बारे में पढ़ रहा था और मैं एक विशेष परिदृश्य के बारे में सोच रहा था। आइए मान लें कि हमारे पास कुछ फाइलें हैं (फ़ाइल ए, फाइलबी, उदाहरण के लिए फाइलसी), प्रत्येक में एकाधिक पूर्णांक शामिल हैं। अगर हम इस तरह कुछ बनाने के लिए सभी फाइलों से संख्याओं को सॉर्ट करना चाहते हैं:

23 fileA 
34 fileB 
35 fileA 
60 fileA 
60 fileC 

नक्शा और प्रक्रिया को कैसे कम किया जाए?

वर्तमान में, मेरे पास यही है लेकिन यह काफी सही नहीं है;

1. (fileName, fileContent) -> (map to) (Number, fileName)

2. अस्थायी कुंजी, मान युग्म तरह और (Number, (list of){fileName1, fileName2...})

3. अस्थायी जोड़े कम करें और मिल

   (Number, fileName1) 
   (Number, fileName2) 
   

   और इतने पर मिल

समस्या यह है कि सॉर्टिंग चरण के दौरान, फ़ाइल नाम वर्णमाला क्रम में नहीं हो सकते हैं और इसलिए कम भाग एक सही आउटपुट उत्पन्न नहीं करेगा। क्या कोई इस परिदृश्य के लिए सही दृष्टिकोण के रूप में कुछ अंतर्दृष्टि प्रदान कर सकता है?

Answer 1

इसे प्राप्त करने का सबसे अच्छा तरीका द्वितीयक प्रकार के माध्यम से है। आपको दोनों कुंजी (आपके केस नंबरों) और मानों को क्रमबद्ध करने की आवश्यकता है (आपके केस फ़ाइल नामों में)। हैडोप में, मैपर आउटपुट केवल चाबियों पर क्रमबद्ध होता है।

यह एक समग्र कुंजी का उपयोग करके हासिल किया जा सकता है: कुंजी जो दोनों संख्याओं और फ़ाइल नामों का संयोजन है। उदाहरण के लिए पहले रिकॉर्ड के लिए, कुंजी (23) के बजाय, (23, fileA) होगी।

आप यहाँ माध्यमिक तरह के बारे में पढ़ सकते हैं: https://www.safaribooksonline.com/library/view/data-algorithms/9781491906170/ch01.html

तुम भी अनुभाग "माध्यमिक क्रमबद्ध", "Hadoop निश्चित गाइड" पुस्तक में के माध्यम से जाना जा सकता है।

सादगी के लिए, मैंने इसे प्राप्त करने के लिए एक कार्यक्रम लिखा है।

इस कार्यक्रम में, कुंजी को मैपर्स द्वारा डिफ़ॉल्ट रूप से क्रमबद्ध किया जाता है। मैंने reducer पक्ष में मूल्यों को क्रमबद्ध करने के लिए एक तर्क लिखा है। तो यह दोनों कुंजी और मूल्यों को क्रमबद्ध करने का ख्याल रखता है और वांछित आउटपुट उत्पन्न करता है।

package com.myorg.hadooptests; 

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 

import java.io.IOException; 
import java.util.*; 

public class SortedValue { 


    public static class SortedValueMapper 
      extends Mapper<LongWritable, Text , Text, IntWritable>{ 

     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 

      String[] tokens = value.toString().split(" "); 

      if(tokens.length == 2) { 
       context.write(new Text(tokens[1]), new IntWritable(Integer.parseInt(tokens[0]))); 
      } 
     } 
    } 

    public static class SortedValueReducer 
      extends Reducer<Text, IntWritable, IntWritable, Text> { 

     Map<String, ArrayList<Integer>> valueMap = new HashMap<String, ArrayList<Integer>>(); 

     public void reduce(Text key, Iterable<IntWritable> values, 
          Context context) throws IOException, InterruptedException { 

      String keyStr = key.toString(); 
      ArrayList<Integer> storedValues = valueMap.get(keyStr); 

      for (IntWritable value : values) { 
       if (storedValues == null) { 
        storedValues = new ArrayList<Integer>(); 
        valueMap.put(keyStr, storedValues); 
       } 
       storedValues.add(value.get()); 
      } 

      Collections.sort(storedValues); 
      for (Integer val : storedValues) { 
       context.write(new IntWritable(val), key); 
      } 
     } 
    } 

    public static void main(String[] args) throws Exception { 

     Configuration conf = new Configuration(); 

     Job job = Job.getInstance(conf, "CompositeKeyExample"); 
     job.setJarByClass(SortedValue.class); 
     job.setMapperClass(SortedValueMapper.class); 
     job.setReducerClass(SortedValueReducer.class); 

     job.setMapOutputKeyClass(Text.class); 
     job.setMapOutputValueClass(IntWritable.class); 

     job.setOutputKeyClass(IntWritable.class); 
     job.setOutputValueClass(Text.class); 

     FileInputFormat.addInputPath(job, new Path("/in/in1.txt")); 
     FileOutputFormat.setOutputPath(job, new Path("/out/")); 

     System.exit(job.waitForCompletion(true) ? 0:1); 

    } 
} 

मैपर तर्क:

1. प्रत्येक पंक्ति पार्स करता है

   के बाद कार्यक्रम है। मानता है कि कुंजी और मान रिक्त वर्ण ("") से अलग होते हैं।

2. यदि लाइन में 2 टोकन हैं, तो यह निकलता है (फ़ाइल नाम, पूर्णांक मान)। उदाहरण के लिए पहले रिकॉर्ड के लिए, यह उत्सर्जित करता है (fileA, 23)।

प्रसारण तर्क:

1. यह एक HashMap, जहां कुंजी फ़ाइल का नाम है और मूल्य उस फ़ाइल के लिए पूर्णांकों की एक सूची है में (कुंजी, मूल्य) जोड़े रखता है। उदाहरण के लिए fileA के लिए, संग्रहीत मान 23, 34 और 35

2. अंततः, यह किसी विशेष कुंजी के लिए मानों को रेखांकित करता है और प्रत्येक मूल्य उत्सर्जक (मान, कुंजी) को reducer से करता है। उदाहरण के लिए fileA के लिए, रिकॉर्ड उत्पादन कर रहे हैं: (23, fileA), (34, fileA) और (35, fileA)

मैं निम्नलिखित इनपुट के लिए इस कार्यक्रम के भाग गया:

34 fileB 
35 fileA 
60 fileC 
60 fileA 
23 fileA 

मुझे मिल गया निम्नलिखित आउटपुट:

23  fileA 
35  fileA 
60  fileA 
34  fileB 
60  fileC