MapReduce开发

前提 1.yarn-site.xml配置


    yarn.nodemanager.aux-services
    mapreduce_shuffle

2.解决HDFS权限问题

API方式

System.setProperty("HADOOP_USER_NAME","root")

配置方式


	dfs.permissions.enabled
	false

MR执行环境

以WordCount程序为例

1.本地测试环境(IDEA)

本地测试以多线程方式模拟服务器运行，无需开启集群
IDEA内创建Java项目，并导入依赖


    1.8
    2.7.1


    
    
        org.apache.hadoop
        hadoop-hdfs
        ${hadoop.version}
    
    
        org.apache.hadoop
        hadoop-common
        ${hadoop.version}
    
    
        org.apache.hadoop
        hadoop-client
        ${hadoop.version}
    
    
        org.apache.hadoop
        hadoop-mapreduce-client-core
        ${hadoop.version}
    
    
        org.apache.hadoop
        hadoop-mapreduce-client-jobclient
        ${hadoop.version}

编写输入数据在本地，注意程序运行起来输出目录会被删一次，所以最好创建一个新的目录，输出目录不用手动创建，程序运行时会自动创建
编写WorkCount程序代码

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;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordCount {

    
    public static class TokenizerMapper
            extends Mapper {//<前两位表示输入类型<偏移量，一行文本>，后两位表示输出类型<单词，数字>>
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        
        @Override
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            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();

        
        @Override
        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();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "word count");//名字
        job.setJarByClass(WordCount.class);//上线的jar
        job.setMapperClass(TokenizerMapper.class);//mapper
        job.setCombinerClass(IntSumReducer.class);//combine:合并一个节点
        job.setReducerClass(IntSumReducer.class);//reduce:合并不同节点
        job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
        job.setOutputValueClass(IntWritable.class);//输出值的类型，与上面一致
        FileInputFormat.addInputPath(job, new Path("E:\HadoopMRData\input"));//输入目录
        FileOutputFormat.setOutputPath(job, new Path("E:\HadoopMRData\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****
        System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
    }
}

观察E:HadoopMRDataoutput目录下的输出数据

2.直接在服务器使用命令的方式调用

程序的执行过程也在服务器上
IDEA内创建Java项目，并导入依赖


    1.8
    2.7.1


    
    
         org.apache.hadoop
         hadoop-core
         1.2.1
     



    
        
            
                org.apache.maven.plugins
                maven-jar-plugin
                2.4

编写WorkCount程序代码

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;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordCount {

    
    public static class TokenizerMapper
            extends Mapper {//<前两位表示输入类型<偏移量，一行文本>，后两位表示输出类型<单词，数字>>

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

        
        @Override
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            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();

        
        @Override
        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();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "word count");//名字
        job.setJarByClass(WordCount.class);//上线的jar
        job.setMapperClass(TokenizerMapper.class);//mapper
        job.setCombinerClass(IntSumReducer.class);//combine:合并一个节点
        job.setReducerClass(IntSumReducer.class);//reduce:合并不同节点
        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);//启动，0表示正常退出
    }
}

打包
将打好的包放到Linux中
hdfs中创建输入目录，不要创建输出目录否则失败
编写输入数据在hdfs
jar在hadoop上运行，命令格式：hadoop jar xxx.jar 类名输入路径输出路径
web观察http://node3:8088/有任务在进行，运行完后观察输出数据

下篇讲讲MR小案例

MapReduce开发

Java相关栏目本月热门文章