用於字符串轉換的Hadoop MapReduce示例

Question

我在某些文本文件中有大量字符串，並且需要通過此類算法轉換此字符串：將字符串轉換爲小寫字母並刪除所有空格。

你能給我一個實現該算法的Hadoop MapReduce函數的例子嗎？

謝謝。

Answer 1

我嘗試了下面的代碼，並在單行中獲取輸出。

公共類TOUPPER {

public static class textMapper extends Mapper<LongWritable,Text,NullWritable,Text> 
{ 
    Text outvalue=new Text(); 

    public void map(LongWritable key,Text values,Context context) throws IOException, InterruptedException 
    { 
     String token; 
     StringBuffer br=new StringBuffer(); 
     StringTokenizer st=new StringTokenizer(values.toString()); 
     while(st.hasMoreTokens()) 
     { 
      token=st.nextToken(); 
      br.append(token.toUpperCase()); 
     } 
     st=null; 
     outvalue.set(br.toString()); 
     context.write(NullWritable.get(), outvalue); 
     br=null; 

    } 
} 
public static class textReduce extends Reducer<NullWritable,Text,NullWritable,Text> 
{ 
    Text outvale=new Text(); 
    public void reduce(NullWritable key,Iterable<Text> values,Context context) throws IOException, InterruptedException 
    { 
     StringBuffer br=new StringBuffer(); 
     for(Text st:values) 
     { 
      br.append(st.toString()); 
     } 
     outvale.set(br.toString()); 
     context.write(NullWritable.get(), outvale); 
    } 
} 
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    Configuration conf=new Configuration(); 
    @SuppressWarnings("deprecation") 
    Job job=new Job(conf,"touipprr"); 

    job.setJarByClass(toUpper.class); 
    job.setMapperClass(textMapper.class); 
    job.setReducerClass(textReduce.class); 

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

    FileInputFormat.addInputPath(job, new Path(args[0])); 
    FileOutputFormat.setOutputPath(job, new Path(args[1])); 

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




} 

}

Answer 2

時，我與玩弄天的map-reduce，我也有類似的想法，必須有一些練習或技術，通過它我們可以修改記錄中的每個單詞並執行所有的清理工作。
當我們回顧map-reduce的整個算法時，我們有一個map函數，它可以在分隔符的幫助下將傳入的記錄拆分爲標記（也許您會更好地瞭解它們）。現在，讓我們嘗試以描述性的方式處理您給出的問題陳述。
以下是我會努力做到，當我是新來的東西映射簡化：

> I will probably write a map() method which will split the lines for me 
> I will possibly run out of options and write a reduce function 
and somehow will be able to achieve my objective 

上述做法是完全好的，但有一個更好的方法，可以幫助你決定是否不你將需要reduce函數，因此你將有更多的選擇讓你思考，並完全專注於實現你的目標，並考慮優化你的代碼。

在這樣的情況下，你的問題陳述落入，一個班級來解救我：ChainMapper 現在，ChainMapper如何工作？以下幾點需要考慮：

---

- >第一個映射器將從HDFS中讀取文件，按照分隔符分割每行並將標記存儲在上下文中。
- >第二變換器將獲得來自第一變換器的輸出，在這裏你可以做各種字符串相關的操作爲您的業務需要，如加密的文本或改變爲大寫或小寫等
- >作爲第二個映射器結果的操作字符串應再次存儲到上下文中
- >現在，如果您需要使用reducer來完成像wordcount這樣的聚合任務，那就去做吧。

---

我有一段代碼可能效率不高（或者有些代碼可能會覺得它很糟糕），但它可以滿足你的目的，因爲你可能會玩mapreduce。

SplitMapper.java

public class SplitMapper extends Mapper<LongWritable,Text,Text,IntWritable>{ 
    @Override 
    public void map(Object key,Text value,Context context) 
            throws IOException,InterruptedException{ 
     StringTokenizer xs=new StringTokenizer(value.toString()); 
     IntWritable dummyValue=new IntWritable(1); 
     while(xs.hasMoreElements()){ 
      String content=(String)xs.nextElement(); 
      context.write(new Text(content),dummyValue); 
     } 
    } 
} 

LowerCaseMapper。java的

public class LowerCaseMapper extends Mapper<Text,IntWritable,Text,IntWritable>{ 
    @Override 
    public void map(Text key,IntWritable value,Context context) 
             throws IOException,InterruptedException{ 
     String val=key.toString().toLowerCase(); 
     Text newKey=new Text(val); 
     Context.write(newKey,value);  
    } 
} 

因爲我在這裏執行wordcount的，所以我需要一個減速

ChainMapReducer.java

public class ChainMapReducer extends Reducer<Text,IntWritable,Text,IntWritable>{ 
    @Override 
    public void reduce(Text key,Iterable<IntWritable> value,Context context) 
           throws IOException,InterruptedException{ 
     int sum=0; 
     for(IntWritable v:value){ 
      sum+=value.get(); 
     } 
     context.write(key,new IntWritables(sum)); 
    } 
} 

爲了能夠實現chainmapper的概念成功的話，你必須注意駕駛員的每一個細節

DriverClass.java

public class DriverClass extends Configured implements Tool{ 
    static Configuration cf; 
    public int run(String args[]) throws IOException,InterruptedException,ClassNotFoundException{ 
     cf=new Configuration(); 
     Job j=Job.getInstance(cf); 
     //configuration for the first mapper 
     Configuration.splitMapConfig=new Configuration(false); 
     ChainMapper.addMapper(j,SplitMapper.class,Object.class,Text.class,Text.class,IntWritable.class,splitMapConfig); 
     //configuration for the second mapper 
     Configuration.lowerCaseConfig=new Configuration(false); 
     ChainMapper.addMapper(j,LowerCaseMapper.class,Text.class,IntWritable.class,Text.class,IntWritable.class,lowerCaseConfig); 

     j.setJarByClass(DriverClass.class); 
     j.setCombinerClass(ChainMapReducer.class); 
     j.setOutputKeyClass(Text.class); 
     j.setOutputValueClass(IntWritable.class); 

     Path outputPath=new Path(args[1]); 
     FileInputFormat.addInputPath(j,new Path(args[0])); 
     FileOutputFormat.setOutputPath(j,outputPath); 
     outputPath.getFileSystem(cf).delete(outputPath,true); 
    } 
    public static void main(String args[]) throws Exception{ 
     int res=ToolRunner.run(cf,new DriverClass(),args); 
     System.exit(1); 
    } 
} 

驅動程序類是相當多理解只有一個需要觀察的ChainMapper.add(<job-object>,<Map-ClassName>,<Input arguments types>,<configuration-for-the-concerned-mapper>)

我希望的解決方案提供你的目的簽名，請讓我如果您嘗試實施時可能會出現任何問題，請告知。
謝謝！