hadoop mapreduce um Teilstrings unterschiedlicher Länge zu erzeugen

Question

Mit Hadoop mapreduce Ich schreibe Code, um Teilstrings unterschiedlicher Länge zu erhalten. Beispiel gegebener String "ZYXCBA" und Länge 3 (Mit einer Textdatei gebe ich Eingabe als "3 ZYXCBA"). Mein Code muss alle möglichen Strings der Länge 3 ("ZYX", "YXC", "XCB", "CBA"), Länge 4 ("ZYXC", "YXCB", "XCBA") abschließend Länge 5 ("ZYXCB "," YXCBA ").

In Karte Phase habe ich folgende:

key = Länge des Teils Ich mag

value = "ZYXCBA".

So Mapper Ausgang ist

3,"ZYXCBA" 
4,"ZYXCBA" 
5,"ZYXCBA" 

In reduziere ich String nehmen ("ZYXCBA") und die Taste 3 alle Substrings der Länge 3. Das Gleiche gilt für 4,5 tritt zu erhalten. Ergebnisse werden mit einer Zeichenfolge verkettet. So löschte von verringern sollte:

3 "ZYX YXC XCB CBA" 
4 "ZYXC YXCB XCBA" 
5 "ZYXCB YXCBA" 

ich meinen Code leite mit folgenden Befehl ein:

hduser@Ganesh:~/Documents$ hadoop jar Saishingles.jar hadoopshingles.Saishingles Behara/Shingles/input Behara/Shingles/output 

Mein Code ist wie unten dargestellt:

package hadoopshingles; 

import java.io.IOException; 
//import java.util.ArrayList; 

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


public class Saishingles{ 

public static class shinglesmapper extends Mapper<Object, Text, IntWritable, Text>{ 

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

      String str = new String(value.toString()); 
      String[] list = str.split(" "); 
      int x = Integer.parseInt(list[0]); 
      String val = list[1]; 
      int M = val.length(); 
      int X = M-1; 


      for(int z = x; z <= X; z++) 
      { 
       context.write(new IntWritable(z), new Text(val)); 
      } 

     } 

    } 


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


    public void reduce(IntWritable key, Text value, Context context 
      ) throws IOException, InterruptedException { 
     int z = key.get(); 
     String str = new String(value.toString()); 
     int M = str.length(); 
     int Tz = M - z; 
     String newvalue = ""; 
     for(int position = 0; position <= Tz; position++) 
     { 
      newvalue = newvalue + " " + str.substring(position,position + z); 
     } 

     context.write(new IntWritable(z),new Text(newvalue)); 
    } 
} 




public static void main(String[] args) throws Exception { 
     GenericOptionsParser parser = new GenericOptionsParser(args); 
     Configuration conf = parser.getConfiguration(); 
     String[] otherArgs = parser.getRemainingArgs(); 

     if (otherArgs.length != 2) 
     { 
      System.err.println("Usage: Saishingles <inputFile> <outputDir>"); 
      System.exit(2); 
     } 
     Job job = Job.getInstance(conf, "Saishingles"); 
     job.setJarByClass(hadoopshingles.Saishingles.class); 
     job.setMapperClass(shinglesmapper.class); 
     //job.setCombinerClass(shinglesreducer.class); 
     job.setReducerClass(shinglesreducer.class); 
     //job.setMapOutputKeyClass(IntWritable.class); 
     //job.setMapOutputValueClass(Text.class); 
     job.setOutputKeyClass(IntWritable.class); 
     job.setOutputValueClass(Text.class); 
     FileInputFormat.addInputPath(job, new Path(args[0])); 
     FileOutputFormat.setOutputPath(job, new Path(args[1])); 
     System.exit(job.waitForCompletion(true) ? 0 : 1); 

} 

} 

Ausgabe reduzieren statt Zurückkehrend

3 "ZYX YXC XCB CBA" 
4 "ZYXC YXCB XCBA" 
5 "ZYXCB YXCBA" 

es wird zurückgegeben

3 "ZYXCBA" 
4 "ZYXCBA" 
5 "ZYXCBA" 

d. H. Es gibt denselben Ausgang wie Mapper. Ich weiß nicht, warum das passiert. Bitte helfen Sie mir, dies zu lösen und danke im Voraus für die Hilfe;) :) :)

Answer 1

Sie können dies erreichen, ohne sogar Reducer zu laufen. Ihre Map/Reduce-Logik ist falsch ... die Umwandlung sollte in Mapper erfolgen.

Reduce - In dieser Phase wird die reduce(WritableComparable, Iterator, OutputCollector, Reporter) Verfahren wird für jedes Paar <key, (list of values)> in den gruppierten Eingängen bezeichnet.

in Ihrer reduce Signatur: public void reduce(IntWritable key, Text value, Context context)

sollte public void reduce(IntWritable key, Iterable<Text> values, Context context)

Auch ändern letzte Zeile reduzieren Methode: context.write(new IntWritable(z),new Text(newvalue)); zu context.write(key,new Text(newvalue)); - Sie bereits Intwritable Key von Mapper haben, kann ich nicht new erstellen würde.

mit gegebenem Eingang:

3 "ZYXCBA" 
4 "ZYXCBA" 
5 "ZYXCBA" 

Mapper Job folgende Ausgabe:

3 "XCB YXC ZYX" 
4 "XCBA YXCB ZYXC" 
5 "YXCBA ZYXCB" 

MapReduceJob:

import java.io.IOException; 
import java.util.ArrayList; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.Reducer.Context; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 

public class SubStrings{ 

    public static class SubStringsMapper extends Mapper<Object, Text, IntWritable, Text> { 

     @Override 
     public void map(Object key, Text value, Context context) throws IOException, InterruptedException { 

      String [] values = value.toString().split(" "); 
      int len = Integer.parseInt(values[0].trim()); 
      String str = values[1].replaceAll("\"", "").trim(); 

      int endindex=len; 
      for(int i = 0; i < len; i++) 
      { 
       endindex=i+len; 
       if(endindex <= str.length()) 
        context.write(new IntWritable(len), new Text(str.substring(i, endindex))); 
      } 

     } 
    } 

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

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

      String str="\""; //adding starting quotes 
      for(Text value: values) 
       str += " " + value; 

      str=str.replace("\" ", "\"") + "\""; //adding ending quotes 
      context.write(key, new Text(str)); 
     } 
    } 

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

     Configuration conf = new Configuration(); 
     Job job = Job.getInstance(conf, "get-possible-strings-by-length"); 

     job.setJarByClass(SubStrings.class); 
     job.setMapperClass(SubStringsMapper.class); 
     job.setReducerClass(SubStringsReducer.class); 

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

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

     FileInputFormat.addInputPath(job, new Path(args[0])); 
     FileOutputFormat.setOutputPath(job, new Path(args[1])); 
     FileSystem fs = null; 
     Path dstFilePath = new Path(args[1]); 
     try { 
      fs = dstFilePath.getFileSystem(conf); 
      if (fs.exists(dstFilePath)) 
       fs.delete(dstFilePath, true); 
     } catch (IOException e1) { 
      e1.printStackTrace(); 
     } 

     job.waitForCompletion(true); 
    } 
}