本例子采用hadoop1.1.2版本，附件中有例子的数据文件
采用气象数据作为处理数据
1、MultipleOutputs例子，具体解释在代码中有注释
|
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
package
StationPatitioner;
import
java.io.IOException;
import
java.util.Iterator;
import
org.apache.hadoop.conf.Configured;
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.mapred.FileInputFormat;
import
org.apache.hadoop.mapred.FileOutputFormat;
import
org.apache.hadoop.mapred.JobClient;
import
org.apache.hadoop.mapred.JobConf;
import
org.apache.hadoop.mapred.MapReduceBase;
import
org.apache.hadoop.mapred.Mapper;
import
org.apache.hadoop.mapred.OutputCollector;
import
org.apache.hadoop.mapred.Reducer;
import
org.apache.hadoop.mapred.Reporter;
import
org.apache.hadoop.mapred.TextOutputFormat;
import
org.apache.hadoop.mapred.lib.MultipleOutputs;
import
org.apache.hadoop.mapred.lib.NullOutputFormat;
import
org.apache.hadoop.util.Tool;
import
org.apache.hadoop.util.ToolRunner;
/**
* hadoop Version 1.1.2
* MultipleOutputs例子
* @author 巧克力黑
*
*/
public
class
PatitionByStationUsingMultipleOutputs
extends
Configured
implements
Tool {
enum
Counter
{
LINESKIP,
//出错的行
}
static
class
StationMapper
extends
MapReduceBase
implements
Mapper<LongWritable , Text, Text , Text>{
private
NcdcRecordParser parser =
new
NcdcRecordParser();
@Override
public
void
map(LongWritable key, Text value,
OutputCollector<Text, Text> output, Reporter reporter)
throws
IOException {
try
{
parser.parse(value);
output.collect(
new
Text(parser.getStationid()), value);
}
catch
(Exception e) {
reporter.getCounter(Counter.LINESKIP).increment(
1
);
//出错令计数器+1
}
}
}
static
class
MultipleOutputReducer
extends
MapReduceBase
implements
Reducer<Text, Text, NullWritable, Text>{
private
MultipleOutputs multipleOutputs;
@Override
public
void
configure(JobConf jobconf) {
multipleOutputs =
new
MultipleOutputs(jobconf);
//初始化一个MultipleOutputs
}
@Override
public
void
reduce(Text key, Iterator<Text> values,
OutputCollector<NullWritable, Text> output, Reporter reporter)
throws
IOException {
//得到OutputCollector
OutputCollector collector = multipleOutputs.getCollector(
"station"
, key.toString().replace(
"-"
,
""
), reporter);
while
(values.hasNext()){
collector.collect(NullWritable.get(), values.next());
//MultipleOutputs用OutputCollector输出数据
}
}
@Override
public
void
close()
throws
IOException {
multipleOutputs.close();
}
}
@Override
public
int
run(String[] as)
throws
Exception {
System.setProperty(
"HADOOP_USER_NAME"
,
"root"
);
//windows下用户与linux用户不一直，采用此方法避免报Permission相关错误
JobConf conf =
new
JobConf();
conf.setMapperClass(StationMapper.
class
);
conf.setReducerClass(MultipleOutputReducer.
class
);
conf.setMapOutputKeyClass(Text.
class
);
conf.setOutputKeyClass(NullWritable.
class
);
conf.setOutputFormat(NullOutputFormat.
class
);
FileInputFormat.setInputPaths(conf,
new
Path(
"hdfs://ubuntu:9000/sample1.txt"
));//input路径
FileOutputFormat.setOutputPath(conf,
new
Path(
"hdfs://ubuntu:9000/temperature"
));//output路径
MultipleOutputs.addMultiNamedOutput(conf,
"station"
, TextOutputFormat.
class
, NullWritable.
class
, Text.
class
);
JobClient.runJob(conf);
return
0
;
}
public
static
void
main(String[] args)
throws
Exception{
int
exitCode = ToolRunner.run(
new
PatitionByStationUsingMultipleOutputs(), args);
System.exit(exitCode);
}
}
|
2、解析气象数据的类
|
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
|
package
StationPatitioner;
import
org.apache.hadoop.io.Text;
public
class
NcdcRecordParser {
private
static
final
int
MISSING_TEMPERATURE =
9999
;
private
String year;
private
int
airTemperature;
private
String quality;
private
String stationid;
public
void
parse(String record) {
stationid = record.substring(
0
,
5
);
year = record.substring(
15
,
19
);
String airTemperatureString;
// Remove leading plus sign as parseInt doesn't like them
if
(record.charAt(
87
) ==
'+'
) {
airTemperatureString = record.substring(
88
,
92
);
}
else
{
airTemperatureString = record.substring(
87
,
92
);
}
airTemperature = Integer.parseInt(airTemperatureString);
quality = record.substring(
92
,
93
);
}
public
String getStationid(){
return
stationid;
}
public
void
parse(Text record) {
parse(record.toString());
}
public
boolean
isValidTemperature() {
return
airTemperature != MISSING_TEMPERATURE && quality.matches(
"[01459]"
);
}
public
String getYear() {
return
year;
}
public
int
getAirTemperature() {
return
airTemperature;
}
}
|
微信关注我们
转载内容版权归作者及来源网站所有！
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。
为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。
Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。
Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。
扫码在手机上查看文章
扫描二维码，手机阅读更方便
有任何问题或合作意向欢迎联系我们
Email: 99873273@qq.com
QQ: 99873273