Olá. Estou enfrentando dificuldades para trabalhar com MapReduce. Sempre que executo a aplicação não consigo resultado, pois aparentemente a função Map executa, porém a função Reduce permanece em 0%.
Quando vou verificar os arquivos que foram gerados no servidor que o hadoop está instalado, o Input esta perfeito, em .txt como definido, porém o reduce gera uma pasta com o nome do arquivo que eu defini (nome_do_arquivo.txt), dentro dessa pasta está presente uma pasta de logs e o arquivo "part-00000", o qual está vazio.
Oque devo fazer para conseguir obter o resultado da operação?
Segue o driver para conectar a aplicação ao Hadoop:
public static void main(String args[]) throws IOException{
System.out.println("Olá");
ControleBD conBD = new ControleBD();
ControleArq conAR = new ControleArq();
conAR.gravar(conBD.pesquisa());
JobConf conf = new JobConf(Principal.class); // Definimos qual classe o job tomará como principal
conf.setJobName("TestePrincipal"); //Nome do job que irá executar na maquina virtual
FileInputFormat.addInputPath(conf, new Path("/user/hadoop-user/input/DadosBancarios.txt")); // Definimos o arquivo de entrada
FileOutputFormat.setOutputPath(conf, new Path("/user/hadoop-user/output/saidaDadosBancarios.txt")); // Definimos o arquivo de saida
conf.setMapperClass(ClasseMapper.class); // configuramos a classe do mapper
conf.setReducerClass(ClasseReducer.class); // configuramos a classe do reducer
conf.setOutputKeyClass(Text.class); // definimos o tipo de saida esperada para as operações de map e reduce, nesse caso Texto
conf.setOutputValueClass(IntWritable.class); // definimos o tipo de saida esperada para as operações de map e reduce, nesse caso inteiros
JobClient.runJob(conf); // Executa o Job com as configurações passadas
}
Segue a classe Mapper:
public class ClasseMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {
public void map(LongWritable chave, Text valor,OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOException {
String linha = valor.toString();
System.out.println(linha);
String ano = "";
int valorIndice = 0;
if(linha.contains("year:")){
String[] divisor = linha.split(":");
ano = divisor[1];
}
if(linha.contains("value:")){
String[] divisor = linha.split(":");
valorIndice = Integer.parseInt(divisor[1]);
}
output.collect(new Text(ano), new IntWritable(valorIndice));
}
}
Segue a classe do Reducer:
public class ClasseReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text chave, Iterator<IntWritable> valor, OutputCollector<Text, IntWritable> output, Reporter reporter)throws IOException {
int maxValue = 99999999;
while (valor.hasNext()) {
maxValue = Math.max(maxValue, valor.next().get());
}
output.collect(chave, new IntWritable(maxValue));
}
}
Segue o log gerado pelo Eclipse-Plugin:
15/09/17 17:13:37 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
15/09/17 17:13:38 INFO mapred.FileInputFormat: Total input paths to process : 1
15/09/17 17:13:38 INFO mapred.FileInputFormat: Total input paths to process : 1
15/09/17 17:13:39 INFO mapred.JobClient: Running job: job_201509170444_0002
15/09/17 17:13:40 INFO mapred.JobClient: map 0% reduce 0%
15/09/17 17:13:48 INFO mapred.JobClient: map 100% reduce 0%
15/09/17 17:13:53 INFO mapred.JobClient: Job complete: job_201509170444_0002
15/09/17 17:13:53 INFO mapred.JobClient: Counters: 16
15/09/17 17:13:53 INFO mapred.JobClient: File Systems
15/09/17 17:13:53 INFO mapred.JobClient: HDFS bytes read=152753
15/09/17 17:13:53 INFO mapred.JobClient: HDFS bytes written=10
15/09/17 17:13:53 INFO mapred.JobClient: Local bytes read=44044
15/09/17 17:13:53 INFO mapred.JobClient: Local bytes written=88160
15/09/17 17:13:53 INFO mapred.JobClient: Job Counters
15/09/17 17:13:53 INFO mapred.JobClient: Launched reduce tasks=1
15/09/17 17:13:53 INFO mapred.JobClient: Launched map tasks=2
15/09/17 17:13:53 INFO mapred.JobClient: Data-local map tasks=2
15/09/17 17:13:53 INFO mapred.JobClient: Map-Reduce Framework
15/09/17 17:13:53 INFO mapred.JobClient: Reduce input groups=1
15/09/17 17:13:53 INFO mapred.JobClient: Combine output records=0
15/09/17 17:13:53 INFO mapred.JobClient: Map input records=6240
15/09/17 17:13:53 INFO mapred.JobClient: Reduce output records=1
15/09/17 17:13:53 INFO mapred.JobClient: Map output bytes=31200
15/09/17 17:13:53 INFO mapred.JobClient: Map input bytes=149856
15/09/17 17:13:53 INFO mapred.JobClient: Combine input records=0
15/09/17 17:13:53 INFO mapred.JobClient: Map output records=6240
15/09/17 17:13:53 INFO mapred.JobClient: Reduce input records=6240
Ps: Não estou utilizando o Hadoop diretamente no meu computador, ele está sendo executado em uma máquina virtual ( https://developer.yahoo.com/hadoop/tutorial/)
Grato
Rafael Muniz