我有一組來自映射器的減速器輸入：(1939, [121, 79, 83, 28]) (1980, [0, 211, ?113])我想得到如下輸出：1939 max:121 min:28 avg: 77.75如果我在我的 reducer 類中不使用如下自定義可寫，我可以得到它：public static class MaxTemperatureReducer extends Reducer<Text, IntWritable, Text, Text> { Text yearlyValue = new Text(); @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; int CounterForAvg = 0; int minValue = Integer.MAX_VALUE; int maxValue = Integer.MIN_VALUE; float avg; for (IntWritable val : values) { int currentValue = val.get(); sum += currentValue; CounterForAvg++; minValue = Math.min(minValue, currentValue); maxValue = Math.max(maxValue, currentValue); } avg = sum / CounterForAvg; String requiredValue = "max temp:"+maxValue + "\t" +"avg temp: "+ avg + "\t"+ "min temp: " +minValue; yearlyValue.set(requiredValue); context.write(key, yearlyValue); } }但是，使用 customwritable 類會產生以下結果：1939 1211939 791939 831939 281980 01980 2111980 -113這是我如何實現自定義類和減速器。我將可迭代對象發送到自定義類并在那里執行計算。我無法弄清楚我在這里做錯了什么。我在 Java 中有 0 exp。
1 回答
慕容3067478
TA貢獻1773條經驗 獲得超3個贊
合并的調用不應該幫助我連接值嗎
當然可以，但你沒有正確使用它。out從未初始化。
CompositeWritable out; // null here
Text textYearlyValue = new Text();
public void reduce(Text key, Iterable<IntWritable> values,
Context context)
throws IOException, InterruptedException {
out.merge(values); // still null, should throw an exception
如果你想輸出一行字符串，你可以只使用一個Text對象。你的merge(Iterable<IntWritable> values)方法可以去任何地方，它不必在一個完全獨立的類中來返回一個 Writable 對象。
但無論如何，如果練習是為了學習如何實現自定義可寫，那么就可以了。
注意事項：
如果你想“組合”多個字段，那么你應該聲明它們
readFields并且write需要按照相同的順序toString確定您在使用TextOutputFormat（默認）時在減速器輸出中看到的內容equals并且hashCode為了完整性而添加（理想情況下你會實現WritableComparable，但這真的只對鍵重要，而不是那么多值）為了與其他 Writables 類似，我將您的
merge方法重命名為set.
你可以期待下面的輸出看起來像
1939 MinMaxAvgWritable{min=28, max=121, avg=77.75}
1980 MinMaxAvgWritable{min=-113, max=211, avg=32.67}
public class MinMaxAvgWritable implements Writable {
private int min, max;
private double avg;
private DecimalFormat df = new DecimalFormat("#.00");
@Override
public String toString() {
return "MinMaxAvgWritable{" +
"min=" + min +
", max=" + max +
", avg=" + df.format(avg) +
'}';
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
MinMaxAvgWritable that = (MinMaxAvgWritable) o;
return min == that.min &&
max == that.max &&
avg == that.avg;
}
@Override
public int hashCode() {
return Objects.hash(min, max, avg);
}
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeInt(min);
dataOutput.writeInt(max);
dataOutput.writeDouble(avg);
}
@Override
public void readFields(DataInput dataInput) throws IOException {
this.min = dataInput.readInt();
this.max = dataInput.readInt();
this.avg = dataInput.readDouble();
}
public void set(int min, int max, double avg) {
this.min = min;
this.max = max;
this.avg = avg;
}
public void set(Iterable<IntWritable> values) {
this.min = Integer.MAX_VALUE;
this.max = Integer.MIN_VALUE;
int sum = 0;
int count = 0;
for (IntWritable iw : values) {
int i = iw.get();
if (i < this.min) this.min = i;
if (i > max) this.max = i;
sum += i;
count++;
}
this.avg = count < 1 ? sum : (sum / (1.0*count));
}
}
有了這個，reducer就很簡單了
public class CompositeReducer extends Reducer<Text, IntWritable, Text, MinMaxAvgWritable> {
private final MinMaxAvgWritable output = new MinMaxAvgWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
// This 'set/merge' method could just as easily be defined here, and return a String to be set on a Text object
output.set(values);
context.write(key, output);
}
}
工作是這樣設置的
// outputs for mapper and reducer
job.setOutputKeyClass(Text.class);
// setup mapper
job.setMapperClass(TokenizerMapper.class); // Replace with your mapper
job.setMapOutputValueClass(IntWritable.class);
// setup reducer
job.setReducerClass(CompositeReducer.class);
job.setOutputValueClass(MinMaxAvgWritable.class); // notice custom writable
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
return job.waitForCompletion(true) ? 0 : 1;
添加回答
舉報
0/150
提交
取消