本例子采用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业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。
为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。
Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。
Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。
扫码在手机上查看文章
扫描二维码，手机阅读更方便
有任何问题或合作意向欢迎联系我们
Email: 99873273@qq.com
QQ: 99873273