YARN集群的mapreduce测试（二）

只有mapTask任务没有reduceTask的情况：

测试准备:

首先同步时间，然后 开启hdfs集群，开启yarn集群；在本地"/home/hadoop/test/"目录创建user文件；

user是存了数据的文件，内容如下：

（导入hadoop-2.7.3-All.jar包）

 测试目标：

 让groupId排在数据的第一列；

测试代码：

outValue = NullWritable.get();//其内部定义决定了只能调用get方法初始化

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class OnlyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    
    private Text outKey;
    private NullWritable outValue;
    
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        outKey = new Text();
        outValue = NullWritable.get();
    }
    
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split("\s+");
//        String groupId = fields[2];
//        StringBuilder builder = new StringBuilder();
//        builder.append(groupId).append("	").append(fields[0]).append("	").append(fields[1]);
        StringBuilder builder = new StringBuilder(fields[2]).append("	").append(fields[0]).append("	").append(fields[1]);
        outKey.set(builder.toString());
        context.write(outKey , outValue);
    }

    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        outKey = null;
    }
    
}

package com.mmzs.bigdata.yarn.mapreduce;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;

/**
 * @author hadoop
 *
 */
public class OnlyDriver {

    private static FileSystem fs;
    private static Configuration conf;
    static {
        String uri = "hdfs://master01:9000/";
        conf = new Configuration();
        try {
            fs = FileSystem.get(new URI(uri), conf, "hadoop");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Job wcJob = getJob(args);
        if (null == wcJob) {
            return;
        }
        //提交Job到集群并等待Job运行完成，参数true表示将Job运行时的状态信息返回到客户端
        boolean flag = false;
        flag = wcJob.waitForCompletion(true);
        System.exit(flag?0:1);
    }
    
    /**
     * 获取Job实例
     * @param args
     * @return
     * @throws IOException
     */
    public static Job getJob(String[] args) throws IOException {
        if (null==args || args.length<2) return null;
        Path localPath = new Path("/home/hadoop/test/user");
        //放置需要处理的数据所在的HDFS路径
        Path inputPath = new Path(args[0]);
        //放置Job作业执行完成之后其处理结果的输出路径
        Path outputPath = new Path(args[1]);

        //如果输入的集群路径不存在，则创建并且将数据文件拷贝到创建的集群路径
        if (!fs.exists(inputPath)) {
            fs.mkdirs(inputPath);
            fs.copyFromLocalFile(localPath, inputPath);
        }
        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);//true表示递归删除
        }
        
        //获取Job实例
        Job wcJob = Job.getInstance(conf, "OnlyMapperJob");
        //设置运行此jar包入口类
        //wcJob的入口是WordCountDriver类
        wcJob.setJarByClass(OnlyDriver.class);
        //设置Job调用的Mapper类
        wcJob.setMapperClass(OnlyMapper.class);
        
        //设置MapTask的输出键类型
        wcJob.setMapOutputKeyClass(Text.class);
        //设置MapTask的输出值类型
        wcJob.setMapOutputValueClass(NullWritable.class);
        
        //如果没有ReduceTask则下面这条语句必须存在
        wcJob.setNumReduceTasks(0);
        
        //设置整个Job需要处理数据的输入路径
        FileInputFormat.setInputPaths(wcJob, inputPath);
        //设置整个Job计算结果的输出路径
        FileOutputFormat.setOutputPath(wcJob, outputPath);
        
        return wcJob;
    }
    
}

测试结果：

运行时传入参数是：

如果在eclipse上运行：传参需要加上集群的master的uri即 hdfs://master01:9000

输入路径参数：  /data/onlymapper/src

输出路径参数：  /data/onlymapper/dst

 

//如果没有ReduceTask则下面这条语句必须存在
wcJob.setNumReduceTasks(0);

输出目录中的结果有：

_SUCCESS（我猜是一种Job运行成功的标志文件，其没有内容）

part-m-00000(只有mapTask任务时就是m)

part-r-00000（既有mapTask任务也有reduceTask任务时就是r）