用Maven构建项目&写一个简单的Mapper-Reducer

开发环境：Windows系统+Eclipse+Hadoop（2.2.0）集群（在虚拟机上）

用到的工具：

[code] SecureCRT   用来连接Hadoop集群
 SecureFX    用来Windows与hadoop集群文件传输
 Notepad++   用来文本编辑，它可以连接hadoop集群中的某个节点，连接成功后可以在本地直接编辑hadoop集群中的文件

构建Maven项目

1. 在Eclipse中新建Maven项目，目录结构如下（如果Eclipse中没有Maven，下载Maven包配置即可，配置方法不做介绍）





2. 目录介绍

src/main/java 存放java源程序

src/main/resources 存放资源文件，例如property文件

src/test/java 存放测试的源文件，比如单元测试之类的

src/test/resoutces 存放测试文件用到的资源文件

根下面的pom.xml文件 定义项目的G***（groupId、artifactId、version），管理依赖等。

简单的Mapper-Reducer流程

1、在src/main/java建立四个包，main存放入口文件，map存放自己写mapper，reduce存放自己写reducer，model存放JavaBean

2、我写的这个Mapper-Reducer主要是用来简单过滤用户信息的数据文件，先看需要过滤的数据文件，这个文件是从http://www.last.fm/中获取到的数据，下图是部分文件信息，





每一列数据的分隔符为“\t”

第一列：用户唯一ID，是用MD5加密过的

第二列：用户的性别（m代表女，f代表男）

第三列：用户年龄

第四列：用户所在国家

第五列：用户注册日期

要达到的目的：我所需要的是第一列、第二列、第四列、第五列，也就是说将第三列去掉，如果文件数量很小，可以手动去删除，但是我所要处理的这个文件用1.5G这么大，想要单纯的打开这个文件都是个问题，所以这里用hadoop集群来处理。

3、问题描述清楚了，现在开始写Mapper-Reducer

在pom.xml中导入开发所需用的jar包，也就是依赖，jar包在你写好依赖以后保存，Maven会自动从网上下载所需用的jar包（需联网，如果已下载Maven所需的仓库，配置即可，具体配置这里不做介绍）

[code]<dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.2.0</version>
</dependency> 
<dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-core</artifactId>
        <version>2.2.0</version>
</dependency>

在map包下新建UserMapper类，继承Mapper类，重写Mapper()方法，具体的业务逻辑就是在这里写的，这里需要注意的是在引入包的时候要引用

import org.apache.hadoop.mapreduce.Mapper;

具体代码如下：

[code]public class UserMapper extends Mapper<LongWritable, Text, Text, User> {

    /**
     *  @param key 行号，我在写代码的时候这块写成了Text，在运行的时候出现类型转换错误的提示，后来才知道这块是行号，是个LongWritable类型
     *  @param value 每一行的具体内容
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //接受每行数据并已"\t"分隔，取出自己所需要的数据，封装到JavaBean里面
        String line = value.toString();
        String[] fields = line.split("\t");
        String userId = fields[0];
        String country = fields[3];
        String date = fields[4];
        User user = new User(userId, country, date);
        context.write(new Text(userId), user);
    }
}

[code]在model包下写JavaBean，需要实现序列化接口Writable，并实现write和readFiles方法，应注意属性序列化的顺序和类型与反序列化的顺序和类型必须保持一致，String的序列化用out.WriteUTF(String s)，代码如下：

[code]    public class User implements Writable {

    private String userId;
    private String country;
    private String date;

    public User() {
    }
    public User(String userId, String country, String date) {
        super();
        this.userId = userId;
        this.country = country;
        this.date = date;
    }
    /**
     * 反序列化
     */
    public void readFields(DataInput in) throws IOException {
        userId = in.readUTF();
        country = in.readUTF();
        date = in.readUTF();
    }
    /**
     * 序列化
     */
    public void write(DataOutput out) throws IOException {
        out.writeUTF(userId);
        out.writeUTF(country);
        out.writeUTF(date);
    }

    /**
    *reducer输出的格式就是toString定义的格式
    */
    public String toString() {
        return country + "\t" + date;
    }

    getter、setter方法.....
}

在reduce包下创建UserReducer，继承Reducer类，重写reduce(）方法,代码如下：

[code]public class UserReducer extends Reducer<Text, User, Text, User> {
    /**
     * @param key map中输出的key
     * @param values map中输出的JavaBean
     */
    @Override
    protected void reduce(Text key, Iterable<User> values, Context context) throws IOException, InterruptedException {
        String userId = "";
        String country = "";
        String date = "";
        for (User user : values) {
            userId = user.getuserId();
            country = user.getcountry();
            date = user.getDate();
        }
        User u = new User(userId, country, date);
        context.write(key, u);
    }
}

在main包下写入口程序User，代码如下：

[code]public class User{
    public static void main(String[] args) throws Exception {
        //用于读取hadoop配置文件的信息
        Configuration conf = new Configuration();
        //job用来加载、配置mapper和reducer，设置输入文件位置和输出文件位置
        Job job = Job.getInstance(conf);
        job.setJarByClass(User.class);
        job.setMapperClass(UserMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(User.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));

        job.setReducerClass(UserReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(User.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //显示执行过程
        job.waitForCompletion(true);
    }
}

最后，打包并放到hadoop集群中namenode为active的节点上（我将打好的jar包放到namenode为standy by的节点上运行，出现了异常，我猜想是运行的时候必须放到active的节点上）

打包：右击项目 -> Export -> Java -> JAR file -> Next -> 选择打包后jar的路径 -> Finish

将打好的jar包，比如user.jar用SecureFX上传到hadoop集群上，

在集群上运行命令：hadoop jar jar文件 入口点 输入文件 输出文件，




注意：原文件和目标文件的位置都指的是hadoop集群中HDFS上的文件位置，源文件在执行前事先放到HDFS中，上传文件代码

hadoop -put /root/user.tsv /user.tsv

,put是上传的意思，后面的文件路径是Linux下的，最后的是HDFS上的路径（根路径是“/”）

当然，也可以在Eclipse中直接远程执行，但是刚开始学习hadoop，远程执行还没研究懂， 后续再进行介绍这部分内容。

一个简单的Mapper-Reducer程序就完成了，运行过程就不给大家截图了，我把最后我从HDFS上下载下来的结果文件的部分信息截图给大家



第一次写博客，仅仅是为了记录自己在学习Hadoop中所做的简单工作，并分享给大家，如果文章中有什么问题，大家可以相互交流，也希望看到这篇文章的大神给我点建议，谢谢。