MapReduce开发

前提

1.yarn-site.xml配置

1. <property>
2.     <name>yarn.nodemanager.aux-services</name>
3.     <value>mapreduce_shuffle</value>
4. </property>

复制代码

2.解决HDFS权限问题

• API方式
  

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

复制代码

• 配置方式
  

1. <property>
2.         <name>dfs.permissions.enabled</name>
3.         <value>false</value>
4. </property>

复制代码

MR执行环境

• 以WordCount程序为例
  

1.本地测试环境(IDEA)

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

1. <properties>
2.     <java.version>1.8</java.version>
3.     <hadoop.version>2.7.1</hadoop.version>
4. </properties>
5. <dependencies>
6.    
7.     <dependency>
8.         <groupId>org.apache.hadoop</groupId>
9.         <artifactId>hadoop-hdfs</artifactId>
10.         <version>${hadoop.version}</version>
11.     </dependency>
12.     <dependency>
13.         <groupId>org.apache.hadoop</groupId>
14.         <artifactId>hadoop-common</artifactId>
15.         <version>${hadoop.version}</version>
16.     </dependency>
17.     <dependency>
18.         <groupId>org.apache.hadoop</groupId>
19.         <artifactId>hadoop-client</artifactId>
20.         <version>${hadoop.version}</version>
21.     </dependency>
22.     <dependency>
23.         <groupId>org.apache.hadoop</groupId>
24.         <artifactId>hadoop-mapreduce-client-core</artifactId>
25.         <version>${hadoop.version}</version>
26.     </dependency>
27.     <dependency>
28.         <groupId>org.apache.hadoop</groupId>
29.         <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
30.         <version>${hadoop.version}</version>
31.     </dependency>
32. </dependencies>

复制代码

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

1. import org.apache.hadoop.conf.Configuration;
2. import org.apache.hadoop.fs.Path;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Job;
6. import org.apache.hadoop.mapreduce.Mapper;
7. import org.apache.hadoop.mapreduce.Reducer;
8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
11. import java.io.IOException;
12. import java.util.StringTokenizer;
14. public class WordCount {
16.     /**
17.      * map映射器：将输入<字节偏移量，一行文本>切分成<单词，次数>
18.      * map之前有一步split，将文本切分成<字节偏移量，一行文本>
19.      */
20.     public static class TokenizerMapper
21.             extends Mapper<Object, Text, Text, IntWritable> {//<前两位表示输入类型<偏移量，一行文本>，后两位表示输出类型<单词，数字>>
22.         private final static IntWritable one = new IntWritable(1);
23.         private Text word = new Text();
24.         /**
25.          * 前两个参是输入类型
26.          *
27.          * @param key     偏移量
28.          * @param value   一行文本，Text类型可序列化，可比较(WritableComparable接口)
29.          * @param context hadoop运行容器，可取出运行时的环境变量
30.          * @throws IOException
31.          * @throws InterruptedException
32.          */
33.         @Override
34.         public void map(Object key, Text value, Context context
35.         ) throws IOException, InterruptedException {
36.             System.out.println("切分split后数据--偏移量:" + key + "\t值:" + value);
37.             StringTokenizer itr = new StringTokenizer(value.toString());//根据自然分隔符分割
38.             while (itr.hasMoreTokens()) {
39.                 word.set(itr.nextToken());//写入文本对象
40.                 context.write(word, one);//保存出去(单词，数字)
41.             }
42.         }
43.     }
45.     /**
46.      * combiner(单节点合并)和reduce(多节点数据合并)都是对相同键的数据进行规约，使用一个即可
47.      * <前两个泛型表示规约的输入数据来源于map的输出，后两个是规约后的单词与数字>
48.      */
49.     public static class IntSumReducer
50.             extends Reducer<Text, IntWritable, Text, IntWritable> {
51.         private IntWritable result = new IntWritable();
53.         /**
54.          * @param key     单词
55.          * @param values  相同单词对应出现次数的集合
56.          *                类中泛型约束是IntWritable，为什么方法上是Iterable<IntWritable>？
57.          *                因为在统计之前会相同的键做成列表word [1,1,1]，然后在规约word 3
58.          * @param context
59.          * @throws IOException
60.          * @throws InterruptedException
61.          */
62.         @Override
63.         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
64.             int sum = 0;
65.             for (IntWritable val : values) {//对一个键
66.                 sum += val.get();//求和规约
67.             }
68.             result.set(sum);
69.             context.write(key, result);
70.         }
71.     }
73.     public static void main(String[] args) throws Exception {
74.         Configuration conf = new Configuration();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
75.         Job job = Job.getInstance(conf, "word count");//名字
76.         job.setJarByClass(WordCount.class);//上线的jar
77.         job.setMapperClass(TokenizerMapper.class);//mapper
78.         job.setCombinerClass(IntSumReducer.class);//combine:合并一个节点
79.         job.setReducerClass(IntSumReducer.class);//reduce:合并不同节点
80.         job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
81.         job.setOutputValueClass(IntWritable.class);//输出值的类型，与上面一致
82.         FileInputFormat.addInputPath(job, new Path("E:\\HadoopMRData\\input"));//输入目录
83.         FileOutputFormat.setOutputPath(job, new Path("E:\\HadoopMRData\\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****
84.         System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
85.     }
86. }

复制代码

• 观察E:\HadoopMRData\output目录下的输出数据
  

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

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

1. <properties>
2.     <java.version>1.8</java.version>
3.     <hadoop.version>2.7.1</hadoop.version>
4. </properties>
5. <dependencies>
6.    
7.     <dependency>
8.          <groupId>org.apache.hadoop</groupId>
9.          <artifactId>hadoop-core</artifactId>
10.          <version>1.2.1</version>
11.      </dependency>
12. </dependencies>
14. <build>
15.     <pluginManagement>
16.         <plugins>
17.             <plugin>
18.                 <groupId>org.apache.maven.plugins</groupId>
19.                 <artifactId>maven-jar-plugin</artifactId>
20.                 <version>2.4</version>
21.             </plugin>
22.         </plugins>
23.     </pluginManagement>
24. </build>

复制代码

• 编写WorkCount程序代码
  

1. import org.apache.hadoop.conf.Configuration;
2. import org.apache.hadoop.fs.Path;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Job;
6. import org.apache.hadoop.mapreduce.Mapper;
7. import org.apache.hadoop.mapreduce.Reducer;
8. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
9. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
11. import java.io.IOException;
12. import java.util.StringTokenizer;
14. public class WordCount {
16.     /**
17.      * map映射器：将输入<字节偏移量，一行文本>切分成<单词，次数>
18.      * map之前有一步split，将文本切分成<字节偏移量，一行文本>
19.      */
20.     public static class TokenizerMapper
21.             extends Mapper<Object, Text, Text, IntWritable> {//<前两位表示输入类型<偏移量，一行文本>，后两位表示输出类型<单词，数字>>
23.         private final static IntWritable one = new IntWritable(1);
24.         private Text word = new Text();
26.         /**
27.          * 前两个参是输入类型
28.          *
29.          * @param key     偏移量
30.          * @param value   一行文本，Text类型可序列化，可比较(WritableComparable接口)
31.          * @param context hadoop运行容器，可取出运行时的环境变量
32.          * @throws IOException
33.          * @throws InterruptedException
34.          */
35.         @Override
36.         public void map(Object key, Text value, Context context
37.         ) throws IOException, InterruptedException {
38.             System.out.println("切分split后数据--偏移量:" + key + "\t值:" + value);
39.             StringTokenizer itr = new StringTokenizer(value.toString());//根据自然分隔符分割
40.             while (itr.hasMoreTokens()) {
41.                 word.set(itr.nextToken());//写入文本对象
42.                 context.write(word, one);//保存出去(单词，数字)
43.             }
44.         }
45.     }
47.     /**
48.      * combiner(单节点合并)和reduce(多节点数据合并)都是对相同键的数据进行规约
49.      * <前两个泛型表示规约的输入数据来源于map的输出，后两个是规约后的单词与数字>
50.      */
51.     public static class IntSumReducer
52.             extends Reducer<Text, IntWritable, Text, IntWritable> {
53.         private IntWritable result = new IntWritable();
55.         /**
56.          * @param key     单词
57.          * @param values  相同单词对应出现次数的集合
58.          *                类中泛型约束是IntWritable，为什么方法上是Iterable<IntWritable>？
59.          *                因为在统计之前会相同的键做成列表word [1,1,1]，然后在规约word 3
60.          * @param context
61.          * @throws IOException
62.          * @throws InterruptedException
63.          */
64.         @Override
65.         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
66.             int sum = 0;
67.             for (IntWritable val : values) {//对一个键
68.                 sum += val.get();//求和规约
69.             }
70.             result.set(sum);
71.             context.write(key, result);
72.         }
73.     }
75.     public static void main(String[] args) throws Exception {
76.         Configuration conf = new Configuration();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
77.         Job job = Job.getInstance(conf, "word count");//名字
78.         job.setJarByClass(WordCount.class);//上线的jar
79.         job.setMapperClass(TokenizerMapper.class);//mapper
80.         job.setCombinerClass(IntSumReducer.class);//combine:合并一个节点
81.         job.setReducerClass(IntSumReducer.class);//reduce:合并不同节点
82.         job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
83.         job.setOutputValueClass(IntWritable.class);//输出值的类型，与上面一致
84.         FileInputFormat.addInputPath(job, new Path(args[0]));//命令行运行时传入
85.         FileOutputFormat.setOutputPath(job, new Path(args[1]));
86.         System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
87.     }
88. }

复制代码

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

下篇讲讲MR小案例


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！