hadoop集群提交代码

Posted petewell

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了hadoop集群提交代码相关的知识,希望对你有一定的参考价值。

在hadoop集群中,写完了mapreduce并没有完成工作,还需要打jar包,然后将jar提交到集群中。hadoop提供了提交jar的入口。

WordCount是写hadoop mapreduce入门级进程,会写wordcount的话,基本上80%的mapreduce就懂了。

mapreduce分为map过程和reduce过程,用户可以根据自己的业务自定义map过程和reduce过程。

以wordcount为例,要计算文本中单词出现的个数,需要读取文本,并针对单词进行统计。

map过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package com.hadoop.mapreduce;

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

import java.io.IOException;



* Created by Frankie on 2018/1/14.
* KEYIN: 默认情况下,是mr框架所读到的一行文本的起始偏移量, Long 在hadoop中有自己的精简序列化接口,所以不直接使用long, 而用LongWritable
* VALUEIN: 默认情况下,是mr框架所读到的一行文本的内容, String
* KEYOUT: 是用户自定义逻辑处理完成后输出数据中的key, 在此处是单词,String
* VALUEOUT: 是用户自定义逻辑处理完成后输出数据中的vlaue, 在次数是单词次数,Integer
*
*
**/

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

/*
* map 阶段的业务逻辑就写在自定义的Map()方法中
* map task会对每一行输入数据调用一次我们自定义map()方法
* */

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
String line = value.toString();
String[] words = line.split(" ");
for(String word: words)
// 将单词作为key, 将次数1作为value,以便于后续的数据分发,可以根据单词分发,以便于相同单词会用到相同的reduce task
// map task会收集,写在一个文档上
context.write(new Text(word), new IntWritable(1));




reduce过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package com.hadoop.mapreduce;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

/**
* Created by Frankie on 2018/1/14.
*
* KEYIN, VALUEIN 对应 mapper输出的KEYOUT, VALUEOUT类型对应
* KEYOUT, VALUEOUT是自定义reduce逻辑处理结果的输出数据类型
* KEYOUT是单词,
* VALUE是总次数
*/
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>

@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
/*
入参key, 是一组相同单词kv对的key
Context 上下文
* */
int count = 0;

// Iterator<IntWritable> iterator = values.iterator();
// while(iterator.hasNext())
// count += iterator.next().get();
//
//

for( IntWritable value: values)
count += value.get();

context.write(key, new IntWritable(count));



mapreduce过程存在一些问题,比如,

Map task如何进行任务分配?

Reduce task如何进行任务分配?

Map task与 reduce task如何进行衔接?

如果某map task 运行失败,如何处理?

Map task如果都要自己负责数据的分区,很麻烦

为例解决这些问题,需要有个master专门对map reduce进行管理。

在WordCount文档中,有专门对作业进行配置,以及最后将代码提交到客户端。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
package com.hadoop.mapreduce;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
* Created by Frankie on 2018/1/14.
*
* 相当于yarn集群的客户端
* 需要在此封装mr进程的相关运行参数,指定jar包,最后提交给yarn
*/

public class WordCount

public static void main(String[] args) throws Exception
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "wordcount");

// 指定本进程的jar包所在的本地路径
job.setJarByClass(WordCount.class);

// 指定本业务使用的map业务类
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);

// 指定mapper输出数据的kv类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

//指定最终输出的数据的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

//指定job的输入原始文档所在目录
// /data/adult.data
FileInputFormat.setInputPaths(job, new Path(args[1]));

// 指定job的输出结果所在目录
FileOutputFormat.setOutputPath(job, new Path(args[2]));

// // 将job中配置的相关参数,以及job所用的java类所在的Jar包,提交给yarn去运行
// job.submit();

// 提交job配置,一直等待到运行结束
boolean res = job.waitForCompletion(true);
System.exit(res? 0: 1);


代码编辑完成后,对代码进行打包。我们在这里选择不依赖第三方包的打包方式进行打包。

打完包后,将生成的jar包提交到服务器中去。
并执行,

1
leiline@master:~/Documents/hadoop/myJars$ hadoop jar HadoopMapReduce.jar com.hadoop.mapreduce.WordCount /data/adult /data/out

注意,out文档是由进程自动创建的,不需要用户手动去创建。最后,代码执行完毕后,可以在hdfs中看到执行的结果:

1
2
3
Found 2 items
-rw-r--r-- 3 leiline supergroup 0 2018-01-14 19:01 /data/out/_SUCCESS
-rw-r--r-- 3 leiline supergroup 216737 2018-01-14 19:01 /data/out/part-r-00000

原文:大专栏  hadoop集群提交代码


以上是关于hadoop集群提交代码的主要内容,如果未能解决你的问题,请参考以下文章

Eclipse远程提交MapReduce任务到Hadoop集群

将MR作业提交给具有不同ID的Hadoop集群

Eclipse远程提交MapReduce任务到Hadoop集群

本地idea开发mapreduce程序提交到远程hadoop集群执行

本地idea开发mapreduce程序提交到远程hadoop集群执行

Spark集群任务提交