Hadoop уменьшает объединение текущего значения с предыдущими значениями - PullRequest
Новая
       13

Hadoop уменьшает объединение текущего значения с предыдущими значениями

0 голосов
/ 29 марта 2012

У меня есть эта функция уменьшения: 

protected void reduce(Text key, Iterable<SortedMapWritable> values, Context context) throws IOException, InterruptedException {
    StringBuilder strOutput = new StringBuilder();
    double sum = 0, i = 0;
    DoubleWritable val = null;

    SortedMapWritable tmp = values.iterator().next();
    strOutput.append("[");
    Set<WritableComparable> keys = tmp.keySet();
    for (WritableComparable mapKey : keys) {                    
        val = (DoubleWritable)tmp.get(mapKey);
        sum += val.get();
        if(i > 0)
            strOutput.append(",");
        strOutput.append(val.get());
        i++;
    }
    strOutput.append("]");

    context.write(new Text(key.toString()), new Text(strOutput.toString()));
    context.write(new Text(key.toString() + "Med"), new Text(Double.toString(sum/i)));
}

В качестве SortedMapWritable я использовал <LongWritable,DoubleWritable>, как мы можем видеть в этом коде 

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    final Context ctx = context;
    Configuration conf = new Configuration();
    FileSystem hdfs = FileSystem.get(conf); 
    Path srcPath = new Path(hdfs.getWorkingDirectory() + "/" + value);  
    Path dstPath = new Path("/tmp/");       

    hdfs.copyToLocalFile(srcPath, dstPath);

    final StringBuilder errbuf = new StringBuilder();
    final Pcap pcap = Pcap.openOffline(dstPath.toString() + "/" +value, errbuf);
    if (pcap == null) {
        throw new InterruptedException("Impossible create PCAP file");
    }

    final HashMap<Integer,JxtaSocketFlow> dataFlows = new HashMap<Integer,JxtaSocketFlow>();
    final HashMap<Integer,JxtaSocketFlow> ackFlows = new HashMap<Integer,JxtaSocketFlow>();

    generateHalfSocketFlows(errbuf, pcap, dataFlows, ackFlows);
    final Text jxtaPayloadKey = new Text("JXTA_Payload");
    final Text jxtaRelyRtt = new Text("JXTA_Reliability_RTT");

    SortedMapWritable payOutput = new SortedMapWritable();
    SortedMapWritable rttOutput = new SortedMapWritable();

    for (Integer dataFlowKey : dataFlows.keySet()) {
        JxtaSocketFlow dataFlow = dataFlows.get(dataFlowKey);
        JxtaSocketStatistics stats = dataFlow.getJxtaSocketStatistics();

        payOutput.put(new LongWritable(stats.getEndTime()), new DoubleWritable((stats.getPayload())/1024));         
        HashMap<Integer,Long> rtts = stats.getRtts();
        for (Integer num : rtts.keySet()) {
            LongWritable key = new LongWritable(stats.getEndTime() + num);                                                      
            rttOutput.put(key, new DoubleWritable(rtts.get(num)));
        }
    }

    try{
        ctx.write(jxtaPayloadKey, payOutput);
        ctx.write(jxtaRelyRtt, rttOutput);
    }catch(IOException e){
        e.printStackTrace();
    }catch(InterruptedException e){
        e.printStackTrace();
    }
}

В функции уменьшения для каждой клавиши значение было объединено с предыдущими значениями.

Например, правильно, ключи и значения должны быть:

key1 -> {a, b, c} key2 -> {d, e, f}

Но значения были

key1 -> {a, b, c} key2 -> {a, b, c, d, e, f}

Кто-нибудь знает, почему это происходит и как мне этого избежать?

Ответы [ 2 ]

3 голосов
/ 02 апреля 2012

Существует открытая ошибка с hadoop https://issues.apache.org/jira/browse/HADOOP-5454, которая может объяснить возникшую проблему.

В следующем коде row.clear () требуется для предотвращения добавления значений в одинитерация к следующему. 

@Log4j
public class StackOverFlowReducer extends Reducer
{
    public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException
    {
        for (SortedMapWritable row : values)
        {
            log.info(String.format("New Map : %s", Joiner.on(",").join(row.entrySet())));
            row.clear();//https://issues.apache.org/jira/browse/HADOOP-5454
        }
    }
}

Я тестировал обходной путь только в пределах одного ключа.Надеюсь, это поможет.

0 голосов
/ 29 марта 2012

Эта проблема не имеет ничего общего с кодом, который вы представили для вашего редуктора, хотя я бы порекомендовал рефакторинг имен его переменных, чтобы было немного легче понять.

Мы можем только заключить, что ваш Mapper передает эти повторяющиеся значения для каждого текущего ключа.Похоже, это и является причиной вашего дублирования.

...