如果我希望存储迭代器的当前值以在 Reduce 方法中与迭代器的下一个值进行比较，Hadoop 要求我克隆它，而不是简单地将其引用分配给临时变量。
我即将将代码发布到我的减速器。
您将看到两个部分：
- Eclipse中测试的主要方法
- 在Hadoop中执行的reduce方法
你会注意到这两行代码是相同的，除了以下几点：
- main 方法从我硬编码到其中的 ArrayList 中获取 Iterator，而 reduce 方法从 mapper 方法中获取 Iterator。
- main 方法当然不会执行 context.write。
这是两者几乎共享的代码：
MMI currentMMI = null;
MMI previousMMI = null;
UltraAggregation currentAggregation = null;
while (values.hasNext()) {
currentMMI = values.next();
if (currentAggregation == null) {
currentAggregation = new UltraAggregation(currentMMI);
}
if (previousMMI == null) {
//previousMMI = new MMI(currentMMI);
previousMMI = currentMMI;
continue;
}
System.out.println();
System.out.println("currentMMI = " + currentMMI);
System.out.println("previousMMI = " + previousMMI);
System.out.println("equals? " + currentMMI.equals(previousMMI));
System.out.println("==? " + (currentMMI == previousMMI));
System.out.println();
// Business logic goes here and involves a context.write on certain conditions
previousMMI = currentMMI;
}
//final context.write
您会注意到，在每个循环结束时，我将刚刚使用的 MMI（“currentMMI”）的引用设置为对象变量“previousMMI”。然后，在下一个循环中，我将 next() 的引用设置为 currentMMI。当我在 Eclipse 中执行我的 main 方法时，以下查询评估为 false，正如预期的那样：
currentMMI == previousMMI;
currentMMI.equals(previousMMI);
但是，当在 Hadoop 中执行时，currentMMI 和 previousMMI 对于以下两个查询总是评估为 true：
currentMMI == previousMMI;
currentMMI.equals(previousMMI);
只有当我将行更改为时previousMMI = currentMMI，previousMMI = new MMI(currentMMI)他们才会评估为假。（我为 MMI 类创建了一个构造函数，它本质上是对传入参数的浅层克隆）。
为什么在hadoop中使用reducer而不是在main方法中时我必须克隆而不是设置引用？
我现在要复制并粘贴reducer 类，它有两部分：eclipse 测试的main 方法和Hadoop 中实际使用的reduce 方法。
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import com.cisco.webex.hadoop.ultrautility.models.MMI;
import com.cisco.webex.hadoop.ultrautility.models.UltraAggregation;
public class MMIReducer extends Reducer<Text, MMI, Object, UltraAggregation> {
public static void main(String[] args) {
ArrayList<MMI> mmis = new ArrayList<MMI>();
mmis.add(new MMI("961864,1,1,1,D1,10,0,2013-08-02 06:00:00.0,USA,N,N"));
mmis.add(new MMI("961865,1,1,1,D1,10,1,2013-08-02 07:00:00.0,USA,N,N"));
mmis.add(new MMI("961866,1,1,1,D1,10,2,2013-08-02 08:00:00.0,USA,N,N"));
mmis.add(new MMI("961867,1,1,1,D1,10,3,2013-08-02 09:00:00.0,USA,N,N"));
mmis.add(new MMI("961868,1,1,1,D1,10,4,2013-08-02 10:00:00.0,USA,N,N"));
mmis.add(new MMI("961869,1,1,1,D1,10,5,2013-08-02 11:00:00.0,USA,N,N"));
mmis.add(new MMI("961870,1,1,1,D1,10,6,2013-08-02 12:00:00.0,USA,N,N"));
mmis.add(new MMI("961871,1,1,1,D1,10,7,2013-08-02 13:00:00.0,USA,N,N"));
mmis.add(new MMI("961872,1,1,1,D1,10,8,2013-08-02 14:00:00.0,USA,N,N"));
mmis.add(new MMI("961873,1,1,1,D1,10,9,2013-08-02 15:00:00.0,USA,N,N"));
Iterator<MMI> values = mmis.iterator();
MMI currentMMI = null;
MMI previousMMI = null;
UltraAggregation currentAggregation = null;
while (values.hasNext()) {
currentMMI = values.next();
if (currentAggregation == null) {
currentAggregation = new UltraAggregation(currentMMI);
}
if (previousMMI == null) {
//previousMMI = new MMI(currentMMI);
previousMMI = currentMMI;
continue;
}
System.out.println();
System.out.println("currentMMI = " + currentMMI);
System.out.println("previousMMI = " + previousMMI);
System.out.println("equals? " + currentMMI.equals(previousMMI));
System.out.println("==? " + (currentMMI == previousMMI));
System.out.println();
// Business logic goes here and involves a context.write on certain conditions
//previousMMI = new MMI(currentMMI);
/*
* THIS DOESNT CAUSE LOGIC ERRORS IN MAIN METHOD
*/
previousMMI = currentMMI;
}
//context.write(null, currentAggregation);
}
@Override
public void reduce(Text key, Iterable<MMI> vals, Context context) throws IOException, InterruptedException {
Iterator<MMI> values = vals.iterator();
//key = deviceId
MMI currentMMI = null;
MMI previousMMI = null;
UltraAggregation currentAggregation = null;
while (values.hasNext()) {
currentMMI = values.next();
if (currentAggregation == null) {
currentAggregation = new UltraAggregation(currentMMI);
}
if (previousMMI == null) {
System.out.println("PreviousMMI is null, setting previousMMI to current MMI and continuing");
//previousMMI = new MMI(currentMMI);
previousMMI = currentMMI;
continue;
}
System.out.println();
System.out.println("currentMMI = " + currentMMI);
System.out.println("previousMMI = " + previousMMI);
System.out.println("equals? " + currentMMI.equals(previousMMI));
System.out.println("==? " + (currentMMI == previousMMI));
System.out.println();
// Business logic goes here and involves a context.write on certain conditions
//previousMMI = new MMI(currentMMI); //Acts as intended
/*
* THIS CAUSES ERRORS WHEN EXECUTED THROUGH HADOOP
*/
previousMMI = currentMMI; // Causes errors
}
context.write(null, currentAggregation);
}
}
当我在 eclipse 中使用静态值执行main 方法时，这是来自 stdout 的截断结果：
currentMMI = Device Id|D1;Entitlement Tag|10;Device Time|Fri Aug 02 07:00:00 PDT 2013;Uptime|1.0
previousMMI = Device Id|D1;Entitlement Tag|10;Device Time|Fri Aug 02 06:00:00 PDT 2013;Uptime|0.0
equals? false
==? false
currentMMI = Device Id|D1;Entitlement Tag|10;Device Time|Fri Aug 02 08:00:00 PDT 2013;Uptime|2.0
previousMMI = Device Id|D1;Entitlement Tag|10;Device Time|Fri Aug 02 07:00:00 PDT 2013;Uptime|1.0
equals? false
==? false
这是我执行 hadoop jar 时截断的结果：
currentMMI = Device Id|D1;Entitlement Tag|10;Device Time|Sun Aug 04 06:00:00 PDT 2013;Uptime|0.0
previousMMI = Device Id|D1;Entitlement Tag|10;Device Time|Sun Aug 04 06:00:00 PDT 2013;Uptime|0.0
equals? true
==? true
currentMMI = Device Id|D1;Entitlement Tag|10;Device Time|Sun Aug 04 07:00:00 PDT 2013;Uptime|1.0
previousMMI = Device Id|D1;Entitlement Tag|10;Device Time|Sun Aug 04 07:00:00 PDT 2013;Uptime|1.0
equals? true
==? true
为什么我必须为 Hadoop 而不是在 Eclipse 中克隆它？