mapreduce程序的三種運行方式

1，linux系統上直接運行mapper.py和reducer.py程序。

cat tmp.txt | python mapper.py | python reducer.py

2，windows系統上運行mapper.py和reducer.py程序。

python mapper.py < tmp.txt | python reducer.py

運行結果：

帶有排序的：

python mapper.py sort |python reducer.py

排序能提高reducer的效率。

3，hadoop中運行mapper和reducer程序。

運行命令：

bin/hadoop jar wc.jar WordCount /opt/hadoop-2.6.0/input /opt/hadoop-2.6.0/output

#可以自動生成output文件夾，存放分詞統計結果。

wc.jar是java開發了程序後打包的jar包。

WordCount 是程序中的類。這個類實現了詞頻統計的功能。

開發運行在hadoop的程序步驟：

1）開發java程序。

2）程序編譯打包。

$ bin/hadoop com.sun.tools.javac.Main WordCount.java #將WordCount.java編譯成三個.class文件

$ jar cf wc.jar WordCount*.class #將三個.class文件打包成jar文件

3)程序運行。

新建input文件夾，用於存放需要統計的文本。

cd /opt/hadoop-2.6.0
mkdir input

複製hadoop-2.6.0文件夾下的txt文件到input文件夾下。

cp *.txt /opt/hadoop-2.6.0/input

運行：

#自動生成output文件夾，存放分詞統計結果
bin/hadoop jar wc.jar WordCount /opt/hadoop-2.6.0/input /opt/hadoop-2.6.0/output 

4）查看運行結果。

代碼如下：

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;

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.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

public static class TokenizerMapper

extends Mapper

private final static IntWritable one = new IntWritable(1);

private Text word = new Text();

public void map(Object key, Text value, Context context

) throws IOException, InterruptedException {

StringTokenizer itr = new StringTokenizer(value.toString());

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word, one);

}

}

}

public static class IntSumReducer

extends Reducer {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable values,

Context context

) throws IOException, InterruptedException {

int sum = 0;

for (IntWritable val : values) {

sum += val.get();

}

result.set(sum);

context.write(key, result);

}

}

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

Configuration conf = new Configuration();

Job job = Job.getInstance(conf, "word count");

job.setJarByClass(WordCount.class);

job.setMapperClass(TokenizerMapper.class);

job.setCombinerClass(IntSumReducer.class);

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

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

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

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

}

}

閱讀更多 熱忱先生 的文章

關鍵字: MapReduce 運行 Hadoop