我在 Hadoop Java API (1.0.3) 中编写了 Map Reduce 作业。这项工作包括对我的数据中特定字段 (X) 的所有值求和,并创建其他字段的加权分布。
输入:
1 field1_1 field2_1 field3_1 ... fieldX_1
2 field1_2 field2_2 field3_2 ... fieldX_2
3 field1_3 field2_3 field3_3 ... fieldX_3
由于我的数据中的任何行都会发出一对,并且我需要使用单个 reducer 来汇总所有值,因此我想将相同的 Reduce 类设置为 Combiner。
X 总和:
地图输出:
X fieldX_1
X fieldX_2
X fieldX_3
X ...
减少输出:
X fieldX_1+fieldX_2+fieldX_3+...
奇怪的是,combiner/reducer 多次收到相同的密钥:
X [fieldX_1 fieldX_1 fieldX_1 ... fieldX_1]
X [fieldX_2 fieldX_2 fieldX_2 ...]
X [fieldX_3 fieldX_3 fieldX_3 ...]
X ...
我对此很确定,因为我正在登录 stderr,所有这些都传递给 reduce 方法的每次调用以进行调试。
我想添加一个更具体的例子:
数据:
1 field1_1 field2_1 field3_1 ... 10
2 field1_2 field2_2 field3_2 ... 20
3 field1_3 field2_3 field3_3 ... 30
4 field1_1 field2_1 field3_1 ... 10
5 field1_2 field2_2 field3_2 ... 40
6 field1_3 field2_3 field3_3 ... 20
...
地图输出:
X 10
X 20
X 30
X 10
X 40
X 20
减少输入(使用组合器):
X [10 10 10 10]
X [20 20 20]
X [30 30 30 30 30 30 30]
X [40 40]
减少输出(使用组合器):
X 40
X 60
X 210
X 80
X 是一个常量标签(字段名称)。请注意,reducer 是使用相同的键 X 和 X 的相同值的集合调用的,例如 [10 10 10 ...] 或 [30 30 30...]。每个总和将单独输出。我的意思是算法工作正常,但在这个阶段需要一个额外的减少步骤来总结重复项。
真实日志示例:
Nov 06, 2013 8:50:12 AM MYCLASS logInputError
WARNING: REDUCE-INPUT: X,[10.0]
Nov 06, 2013 8:50:12 AM MYCLASS logOutputError
WARNING: REDUCE-OUTPUT: X,10.0
Nov 06, 2013 8:50:12 AM MYCLASS logInputError
WARNING: REDUCE-INPUT: X,[25.865, 25.865]
Nov 06, 2013 8:50:12 AM MYCLAS logOutputError
WARNING: REDUCE-OUTPUT: X,51.73
Nov 06, 2013 8:50:12 AM MYCLASS logInputError
WARNING: REDUCE-INPUT: X,[1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4, 1449271.4]
Nov 06, 2013 8:50:12 AM MYCLASS logOutputError
WARNING: REDUCE-OUTPUT: X,2.0289798E7
Nov 06, 2013 8:50:12 AM MYCLASS logInputError
WARNING: REDUCE-INPUT: X,[514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53, 514994.53]
Nov 06, 2013 8:50:12 AM MYCLASS logOutputError
WARNING: REDUCE-OUTPUT: X,6694929.0
Nov 06, 2013 8:50:12 AM MYCLASS logInputError
WARNING: REDUCE-INPUT: X,[1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5, 1438665.5]
Nov 06, 2013 8:50:12 AM MYCLASS logOutputError
WARNING: REDUCE-OUTPUT: X,1.8702654E7
如果我删除组合器一切正常。我知道Combiner 可能会被调用0、1 或多次,但是Reducer 呢?它应该只被调用一次,不是吗?
但更奇怪的是,我对字段分布重复了类似的过程,而这只发生在 X 求和问题上......
加权场分布
映射输出(示例字段 1):
field1_1 X_1
field1_2 X_2
field1_3 X_3
...
减少输出:
field1(class1) fieldX(class1)+fieldX(class1)+fieldX(class1)+...
field1(class2) fieldX(class2)+fieldX(class2)+fieldX(class2)+...
field1(class3) fieldX(class3)+fieldX(class3)+fieldX(class3)+...
...
本质上,对于 field1 的每个值,我总结了 fieldX 的所有相关值,并对几个字段(field1、field2、field3 ...)重复相同的过程。
对于那些发出的对,reducer 接收单个键 (field1(class1)) 和值数组 ([fieldX(class1)...]) 应该正常运行。
结论
一个考虑因素是,对于 X 求和问题,单个键 (X) 映射的值数量等于数据的大小(行数)。同时,对于字段加权分布,值分布在字段包含的几个类标签中。
这是我的代码中的一个错误,还是我没有考虑 Hadoop 的一些程序细节?
根据 M/R 范式,Reducer 类应该一次接收特定键的所有值,而不是分成更多分区。
希望能收到好的反馈。