mapreduce -- wordcount执行流程

建立pom：

注：要和安装的hadoop版本一致，并且运行hdfs <?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>

<groupId>xxxx.com.test</groupId> <artifactId>WordCount</artifactId> <version>1.0-SNAPSHOT</version>

<name>WordCount</name> <url>http://www.example.com</url>

<properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> </properties>

<dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.6.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.6.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>2.6.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-common</artifactId> <version>2.6.5</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-jobclient</artifactId> <version>2.6.5</version> </dependency> </dependencies> </project>

resource文件：

1.core-site.xml 2.hdfs-site.xml 3.mapred-site.xml 4.yarn-site.xml 5.log4j.properties 直接从conf中拉取即可，log4j用来查看日志信息，如：

map端：

package xxxx.com.test.wc; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException;

public class WordCount_Map extends Mapper<LongWritable,Text, Text, IntWritable> { Text k = new Text(); IntWritable v = new IntWritable(1);

@Override
// 每行都调用map方法 将每行的信息读入·Text value·
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    // 将每行的读入数据转化成String
    String line = value.toString();
    // 按空格切分
    String[] words = line.split(" ");
    for(String word : words){
        k.set(word);
        // 将结果写入缓存区，map阶段将单词拆分，并不合并，所以固定值为1
        context.write(k,v);
    }
}

}

reduce端：

package xxxx.com.test.wc; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; import java.util.Iterator;

public class WordCount_Reduce extends Reducer<Text, IntWritable,Text,IntWritable> { Iterator<IntWritable> it_Int = null; IntWritable value = new IntWritable(0);

@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int count = 0;
    // 通过迭代器进行遍历
    it_Int = values.iterator();
    while (it_Int.hasNext()){
        count += it_Int.next().get();
    }
    value.set(count);
    context.write(key,value);
}

}

driver端：

package xxxx.com.test.wc; 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.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; import java.net.URISyntaxException;

public class WC_Driver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException { // TODO Auto-generated method stub Configuration conf = new Configuration(); // conf.set("mapreduce.app-submission.cross-platform","true"); /FileSystem fs = null; conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem"); fs = FileSystem.get(new URI("hdfs://bd1602:50070"),conf);/ // 1.获取job对象 Job job = Job.getInstance(conf); // 设置job的各种属性 job.setJobName("wordcountjob-"); // 2.设置jar存储位置 job.setJarByClass(WC_Driver.class); // 3.关联map和reduce job.setMapperClass(WordCount_Map.class); job.setReducerClass(WordCount_Reduce.class); // 4.设置map阶段输出的key和value job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5.设置最终数据的输出形式 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6.设置输入输出路径 FileInputFormat.setInputPaths(job, new Path("/bdp/harry.txt")); // FileInputFormat.setInputPaths(job, new Path("./bdp/harry.txt")); FileOutputFormat.setOutputPath(job, new Path("/bdp/wordcount-" + System.currentTimeMillis())); // FileOutputFormat.setOutputPath(job, new Path("./bdp/wordcount-" + System.currentTimeMillis())); // 7.提交 boolean result = job.waitForCompletion(true); System.err.println(result); // System.exit(result ? 0: 1); } }