|
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
package
com.baidu.uilt;
import
java.io.*;
import
org.apache.hadoop.io.*;
public
class
TextPair
implements
WritableComparable<TextPair> {
private
Text first;
private
Text second;
public
TextPair() {
set(
new
Text(),
new
Text());
}
public
TextPair(String first, String second) {
set(
new
Text(first),
new
Text(second));
}
public
TextPair(Text first, Text second) {
set(first, second);
}
public
void
set(Text first, Text second) {
this
.first = first;
this
.second = second;
}
public
Text getFirst() {
return
first;
}
public
Text getSecond() {
return
second;
}
@Override
public
void
write(DataOutput out)
throws
IOException {
first.write(out);
second.write(out);
}
@Override
public
void
readFields(DataInput in)
throws
IOException {
first.readFields(in);
second.readFields(in);
}
@Override
public
int
hashCode() {
return
first.hashCode() *
163
+ second.hashCode();
}
@Override
public
boolean
equals(Object o) {
if
(o
instanceof
TextPair) {
TextPair tp = (TextPair) o;
return
first.equals(tp.first) && second.equals(tp.second);
}
return
false
;
}
@Override
public
String toString() {
return
first +
"\t"
+ second;
}
@Override
public
int
compareTo(TextPair tp) {
int
cmp = first.compareTo(tp.first);
if
(cmp !=
0
) {
return
cmp;
}
return
second.compareTo(tp.second);
}
public
static
class
Comparator
extends
WritableComparator {
private
static
final
Text.Comparator TEXT_COMPARATOR =
new
Text.Comparator();
public
Comparator() {
super
(TextPair.
class
);
}
@Override
public
int
compare(
byte
[] b1,
int
s1,
int
l1,
byte
[] b2,
int
s2,
int
l2) {
try
{
int
firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
int
firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
int
cmp = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
if
(cmp !=
0
) {
return
cmp;
}
return
TEXT_COMPARATOR.compare(b1, s1 + firstL1, l1 - firstL1,
b2, s2 + firstL2, l2 - firstL2);
}
catch
(IOException e) {
throw
new
IllegalArgumentException(e);
}
}
}
static
{
WritableComparator.define(TextPair.
class
,
new
Comparator());
}
public
static
class
FirstComparator
extends
WritableComparator {
private
static
final
Text.Comparator TEXT_COMPARATOR =
new
Text.Comparator();
public
FirstComparator() {
super
(TextPair.
class
);
}
@Override
public
int
compare(
byte
[] b1,
int
s1,
int
l1,
byte
[] b2,
int
s2,
int
l2) {
try
{
int
firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
int
firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
return
TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
}
catch
(IOException e) {
throw
new
IllegalArgumentException(e);
}
}
@Override
public
int
compare(WritableComparable a, WritableComparable b) {
if
(a
instanceof
TextPair && b
instanceof
TextPair) {
return
((TextPair) a).first.compareTo(((TextPair) b).first);
}
return
super
.compare(a, b);
}
}
}
|
|
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
package
com.baidu.loan;
/***
*
* /home/users/ouerqiang/hadoop/hadoop-client-palo/hadoop/bin/hadoop jar LoanIdeaInfoText.jar com.baidu.loan.LoanIdeainfoJoinIterialByDAILI6 /test/fbiz/loan/ideainfo/LoanIdeainfoByDAILIUnitID_0928 /test/fbiz/loan/ideainfo/LoanIterialByDAI_0928 /test/fbiz/loan/ideainfo/LoanIdeainfoJoinIterialByDAILI6_1_0928
*
* **/
import
java.io.IOException;
import
java.util.Iterator;
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.Partitioner;
import
org.apache.hadoop.mapred.Reducer;
import
org.apache.hadoop.mapred.Reporter;
import
org.apache.hadoop.mapred.TextInputFormat;
import
org.apache.hadoop.mapred.lib.MultipleInputs;
import
org.apache.hadoop.util.Tool;
import
org.apache.hadoop.util.ToolRunner;
import
org.apache.hadoop.conf.Configured;
import
org.apache.hadoop.fs.Path;
import
org.apache.hadoop.io.LongWritable;
import
org.apache.hadoop.io.Text;
import
com.baidu.uilt.TextPair;
public
class
LoanIdeainfoJoinIterialByDAILI6
extends
Configured
implements
Tool {
public
static
class
JoinUnitMapper
extends
MapReduceBase
implements
Mapper<LongWritable, Text, TextPair, Text> {
public
void
map(LongWritable key, Text value,
OutputCollector<TextPair, Text> output, Reporter reporter)
throws
IOException {
String gbkStr = value.toString();
if
(gbkStr.split(
"\t"
).length <
2
&& gbkStr.split(
","
).length ==
4
) {
String[] strs = gbkStr.split(
","
);
output.collect(
new
TextPair(strs[
0
],
"0"
), value);
}
}
}
public
static
class
JoinIterialMapper
extends
MapReduceBase
implements
Mapper<LongWritable, Text, TextPair, Text> {
public
void
map(LongWritable key, Text value,
OutputCollector<TextPair, Text> output, Reporter reporter)
throws
IOException {
String gbkStr = value.toString();
if
(gbkStr.split(
"\t"
).length >
4
) {
// LoanIterial
String[] strs = gbkStr.split(
"\t"
);
output.collect(
new
TextPair(strs[
0
],
"1"
), value);
}
}
}
public
static
class
JoinReducer
extends
MapReduceBase
implements
Reducer<TextPair, Text, Text, Text> {
public
void
reduce(TextPair key, Iterator<Text> values,
OutputCollector<Text, Text> output, Reporter reporter)
throws
IOException {
Text stationName =
new
Text(values.next());
while
(values.hasNext()) {
Text record = values.next();
Text outValue =
new
Text(stationName.toString() +
"\t"
+ record.toString());
output.collect(stationName, record);
//output.collect(key.getFirst(), outValue);
}
}
}
public
static
class
KeyPartitioner
implements
Partitioner<TextPair, Text> {
@Override
public
void
configure(JobConf job) {}
@Override
public
int
getPartition(TextPair key, Text value,
int
numPartitions) {
return
(key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;
}
}
@Override
public
int
run(String[] args)
throws
Exception {
if
(args.length !=
3
) {
return
-
1
;
}
JobConf conf =
new
JobConf(getConf(), getClass());
conf.setJobName(
"Join record with station name"
);
String strPathUnit =args[
0
];
String strPathIterial =args[
1
];
Path outputPath=
new
Path(args[
2
]);
MultipleInputs.addInputPath(conf,
new
Path(strPathUnit),
TextInputFormat.
class
, JoinUnitMapper.
class
);
MultipleInputs.addInputPath(conf,
new
Path(strPathIterial),
TextInputFormat.
class
, JoinIterialMapper.
class
);
FileOutputFormat.setOutputPath(conf, outputPath);
conf.setPartitionerClass(KeyPartitioner.
class
);
conf.setOutputValueGroupingComparator(TextPair.FirstComparator.
class
);
conf.setMapOutputKeyClass(TextPair.
class
);
conf.setReducerClass(JoinReducer.
class
);
conf.setOutputKeyClass(Text.
class
);
JobClient.runJob(conf);
return
0
;
}
public
static
void
main(String[] args)
throws
Exception {
int
exitCode = ToolRunner.run(
new
LoanIdeainfoJoinIterialByDAILI6(), args);
System.exit(exitCode);
}
}
|
需要注意的是上面的代码只是针对两表的一对一（多）关系，不满足多对多关系。如果需要满足多对多关系则需要加上一下判断即可。
本文转自 梦朝思夕 51CTO博客，原文链接:http://blog.51cto.com/qiangmzsx/1560553
微信关注我们
转载内容版权归作者及来源网站所有！
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。
马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。
为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。
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