Como utilizar CombineFileInputFormat em Hadoop?
Eu quero usar o CombineFileInputFormat de Hadoop 0.20.0 / 0.20.2 de modo que ele processa 1 arquivo por registro e também não compromete a localidade de dados (que normalmente cuida).
É mencionado no Guia Definitivo de Tom White, mas ele não mostrou como fazê-lo. Em vez disso, ele passa para arquivos de sequência. Estou bastante confuso sobre qual é o Significado de variável processada num leitor de discos. Qualquer exemplo de código seria de grande ajuda. Obrigado antecipadamente..2 answers
Verifique o formato de entrada abaixo usado para combinar o formato de entrada de ficheiros.
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
/**
* CustomInputformat which implements the createRecordReader of abstract class CombineFileInputFormat
*/
public class MyCombineFileInputFormat extends CombineFileInputFormat {
public static class MyRecordReader extends RecordReader<LongWritable,Text>{
private LineRecordReader delegate=null;
private int idx;
public MyRecordReader(CombineFileSplit split,TaskAttemptContext taskcontext ,Integer idx) throws IOException {
this.idx=idx;
delegate = new LineRecordReader();
}
@Override
public void close() throws IOException {
delegate.close();
}
@Override
public float getProgress() {
try {
return delegate.getProgress();
}
catch(Exception e) {
return 0;
}
}
@Override
public void initialize(InputSplit split, TaskAttemptContext taskcontext) throws IOException {
CombineFileSplit csplit=(CombineFileSplit)split;
FileSplit fileSplit = new FileSplit(csplit.getPath(idx), csplit.getOffset(idx), csplit.getLength(idx), csplit.getLocations());
delegate.initialize(fileSplit, taskcontext);
}
@Override
public LongWritable getCurrentKey() throws IOException,
InterruptedException {
return delegate.getCurrentKey();
}
@Override
public Text getCurrentValue() throws IOException, InterruptedException {
return delegate.getCurrentValue();
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
return delegate.nextKeyValue();
}
}
@SuppressWarnings("unchecked")
@Override
public RecordReader createRecordReader(InputSplit split,TaskAttemptContext taskcontext) throws IOException {
return new CombineFileRecordReader((CombineFileSplit) split, taskcontext, MyRecordReader.class);
}
}
Aqui está a maneira mais simples de usar CombineFileInputFormat a partir da chamada "nova API". Suponha que seu formato de entrada atual é MyFormat, e ele funciona com chaves de MyKey e valores de MyValue (pode ser alguma subclasse de SequenceFileInputFormat< MyKey, MyValue >
, por exemplo).
public class CombinedMyFormat extends CombineFileInputFormat< MyKey, MyValue > {
// exists merely to fix the key/value types and
// inject the delegate format to the superclass
// if MyFormat does not use state, consider a constant instead
private static class CombineMyKeyMyValueReaderWrapper
extends CombineFileRecordReaderWrapper< MyKey, MyValue > {
protected CombineMyKeyMyValueReaderWrapper(
CombineFileSplit split, TaskAttemptContext ctx, Integer idx
) throws IOException, InterruptedException {
super( new MyFormat(), split, ctx, idx );
}
}
@Override
public RecordReader< MyKey, MyValue > createRecordReader(
InputSplit split, TaskAttemptContext ctx
) throws IOException {
return new CombineFileRecordReader< MyKey, MyValue >(
( CombineFileSplit )split, ctx, CombineMyKeyMyValueReaderWrapper.class
);
}
}
No seu motorista de trabalho, você deve agora ser capaz de apenas cair em CombinedMyFormat
para MyFormat
. Deverá também definir uma propriedadedo tamanho dividido máximo para evitar que o Hadoop combine toda a entrada num único dividir.