大数据实验（四）MapReduce编程实践（Ubuntu)

大数据实验（四）MapReduce编程实践（Ubuntu)

前置工具及环境

Ubuntu 16.4

VirtualBox

Hadoop 2.7.3

jdk1.8

一、MapReduce简介

MapReduce是Hadoop提供的一个分布式计算框架，MapReduce 作业通过将输入的数据集拆分为独立的块，这些块由 map 以并行的方式处理，框架对 map 的输出进行排序，然后输入到 reduce 中。MapReduce 框架专门用于 <key，value> 键值对处理，它将作业的输入视为一组 <key，value> 对，并生成一组 <key，value> 对作为输出。

过程大概如此：

(input) <k1, v1> -> map -> <k2, v2> -> reduce -> <k3, v3> (output)

二、编写Map/Reduce代码

编写用于WordCount统计词条的代码。

000、数据

准备一些文件，用于WordCount统计使用。

我的文件是Hadoop上的：data/123.txt。。。。
内容是一些hadoop hdfs等通过空格分开的词条

001、依赖包

使用Maven构建项目使用以下依赖即可。

或者自行导入。

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop.version}</version>
</dependency>

002、Map代码

public class WordMap extends Mapper<LongWritable, Text,Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //将序列化的类型转换为String
        String s = value.toString();
        //工具类
        StringTokenizer tokenizer = new StringTokenizer(s);
        while (tokenizer.hasMoreTokens()){
            context.write(new Text(tokenizer.nextToken()),new IntWritable(1));
        }

    }
}

003、Reduce代码

public class WordReducer extends Reducer<Text, IntWritable,Text,IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count=0;//计数器

        Iterator<IntWritable> iterator = values.iterator();

        while (iterator.hasNext()){
            count+=iterator.next().get();
        }
        System.out.println("reduce over...");
        context.write(key,new IntWritable(count));

    }
}

004、作业代码

其实很多信息是可以外部输入的或者写在配置文件中便于复用，但我直接设置也是可以的。

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

        System.out.println("word count");

        Configuration conf=new Configuration();

        // 指明 HDFS 的地址
        conf.set("fs.defaultFS", "hdfs://192.168.56.101:9000");

        //设置一个用户，如果当前用户就是hadoop，则可以略去
        System.setProperty("HADOOP_USER_NAME","hadoop");


        //设置job及相关的各种参数
        Job job = Job.getInstance(conf, "wordcount");
        job.setJarByClass(WordCountDriver.class);
        job.setMapperClass(WordMap.class);
        job.setReducerClass(WordReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //作业的输入目录和输出目录，输出目录必须不存在，hadoop会自动创建
        FileInputFormat.addInputPaths(job,"data");
        FileOutputFormat.setOutputPath(job, new Path("dataDir"));

        //设置job执行等待，并打印进度
        boolean b = job.waitForCompletion(true);

        //作业结束
        System.exit(b?0:1);


    }
}

此时已经可以运行，在ide中运行时会出现相关信息。

此时在命令行下查看结果：

#在相应目录下生成几个文件，partxxxx文件是记录结果的文件
hadoop@Hadoop:~$ hadoop fs -cat /user/hadoop/dataDir/part-r-00000
hadoop	2
hbase	1
hive	1
java	1
spark	2

三、打包执行

上面是在ide下运行的，也经常在服务器中运行，此时需要打包。

由于使用Maven构建，直接打包就行。

mvn clean package

由于信息都是写死在java文件中的，所以只需要一个main文件名就可以运行，结果也是一样的。

hadoop jar hadoop-1.0-SNAPSHOT.jar  mapreduce.WordCountDriver