c++ - Hadoop Pipes 1.1.2 でスタックを減らす (java.io.IOException: Bad command code: 48)

Question

おはようございます。

Hadoop Pipes 1.1.2 を実行している完全なクラスターで作業していますが、非常に長いジョブで次のエラーが見つかりました。

2013-07-24 06:13:55,058 ERROR org.apache.hadoop.mapred.pipes.BinaryProtocol: java.io.IOException: Bad command code: 48
    at org.apache.hadoop.mapred.pipes.BinaryProtocol$UplinkReaderThread.run(BinaryProtocol.java:157)

(これは 1 つのノードに 1 つ)

2013-07-24 06:18:03,472 ERROR org.apache.hadoop.mapred.pipes.BinaryProtocol: java.lang.NegativeArraySizeException
    at org.apache.hadoop.mapred.pipes.BinaryProtocol$UplinkReaderThread.readObject(BinaryProtocol.java:180)
    at org.apache.hadoop.mapred.pipes.BinaryProtocol$UplinkReaderThread.run(BinaryProtocol.java:132)

(reduce タスクを実行する残りの 1 つ)

これは私の減速コードです。そこでマップを使用して、文字列内のセカンダリ キーで並べ替えます。パフォーマンスの向上も歓迎します!

void ACEReducer::reduce(HadoopPipes::ReduceContext& context) {
    string fragmento, instante, contenido;
    size_t pos;

    string fichero = string();
    map<string, string> resultados;

    // Ordena la lista de valores para escribirlos conforme al instante simulado (map se encarga de la ordenacon)
    int nValores = 0;
    while (context.nextValue()) {
        fragmento = context.getInputValue();
        pos = fragmento.find("\t", 0);

        if (pos == string::npos) {
            cerr << "ERROR - Valor erroneo en Reduce." << endl;
        } else {
            instante = fragmento.substr(0, pos);
            contenido = fragmento.substr(pos+1, string::npos);

            resultados.insert(pair<string&, string&>(instante, contenido));
        }

        nValores++;
        if (nValores % 10 == 0) {
            context.progress();
        }
    }

    // Concatena los resultados ordenados y los emite para escritura.
    nValores = 0;
    for (map<string, string>::iterator it=resultados.begin(); it!=resultados.end(); it++){
        nValores++;
        if (nValores % 10 == 0) {
            context.progress();
        }
        fichero += it->second;
    }
    context.emit(context.getInputKey(), fichero);
}

IOException をスローしているコードは、Java BinaryProtocol の run() 関数にあります。ここの 120 行目にあります。NegativeArraySizeException は、133 行目と 149 行目の間のどこかでスローされます。私の推測では、int numBytes = WritableUtils.readVInt(inStream);それは INT_MAX よりも高いですが、修正方法がわかりません。

HDFS はすべてのノードで正常に見えますが、使用されているのはわずか 3% です。少なくとも 64 GB の RAM を搭載した非常に堅牢なノードを使用しています。

この問題を検索しましたが、ヒントが見つからなかったので、助けていただければ幸いです。

前もって感謝します、

リオ