gpt4 book ai didi

java - 在 MapReduce 中使用 Custom Partitioner 时清空 reduce 输出

转载 作者:可可西里 更新时间:2023-11-01 14:51:31 24 4
gpt4 key购买 nike

问题陈述:

输入

Monami 45000 A
Tarun 34000 B
Riju 25000 C
Rita 42000 A
Mithun 40000 A
Archana 21000 C
Shovik 32000 B

我想在 Mapreduce 中使用 Custom Partitioner 将 A、B 和 C 级员工记录分隔在三个不同的输出文件中。

输出 1

Monami 45000 A
Rita 42000 A
Mithun 40000 A

输出 2

Tarun 34000 B
Shovik 32000 B

输出 3

Riju 25000 C
Archana 21000 C

map 代码:

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;

public class Map
extends Mapper<LongWritable,Text,Text,Text>
{

//private Text key1 = new Text();
//private Text value1 = new Text();

@Override
protected void map(LongWritable key,Text value,Context context)
throws IOException,InterruptedException
{
String line = value.toString();
String[] part = line.split("\t");
int len = part.length;

//System.out.println(len);
if (len == 3)
{
context.write(new Text(part[2]), new Text(part[0]+"\t"+part[1]));
//System.out.println(part[0]+part[1]+part[2]);
}

}

分区代码

 import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class CustomPartitioner
extends Partitioner<Text,Text>
{
@Override
public int getPartition(Text key, Text value, int numReduceTasks)
{
if(numReduceTasks==0)
return 0;
if(key.equals(new Text("A")))
return 0;
if(key.equals(new Text("B")))
return 1;
else
return 2;
}
}

减少代码

 import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class Reduce
extends Reducer<Text,Text,Text,Text>
{

@Override
protected void reduce(Text key,Iterable<Text> values,Context context)
throws IOException,InterruptedException
{
Iterator<Text> itr = values.iterator();

while(itr.hasNext())
{
context.write(new Text(itr.next().getBytes()),new Text(key));
}


}
}

驱动类

 import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class MapReduceDriver
{
public static void main(String[] args) throws Exception
{
Job job = new Job();

job.setJarByClass(MapReduceDriver.class);
job.setJobName("Custom Partitioner");

FileInputFormat.addInputPath(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));

job.setMapperClass(Map.class);
job.setPartitionerClass(CustomPartitioner.class);
job.setReducerClass(Reduce.class);
job.setNumReduceTasks(3);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

System.exit(job.waitForCompletion(true)?0:1);

}

}

代码运行没有任何错误,但三个 reduce 输出文件为空。此外,当作业运行时,它会将映射输出字节显示为零。因此我相信 map 不会生成任何键值对。但我找不到原因。你能帮我找出错误吗?

我还有一个困惑:在 Map 类中,当检查变量 len > 0 时,我得到 ArrayIndexOutOfBoundsException 但如果使用 == 3 检查,它运行正常,没有任何异常。为什么它会抛出 > 0 的异常吗?

最佳答案

问题是您的输入数据(粘贴在此处)不是制表符分隔的,而是逗号分隔的。如果您替换此行,它应该可以正常工作:

 String[] part = line.split("\t");

用这一行:

String[] part = line.split(" ");

检查 len > 0 时出现异常的原因是您的字符串未拆分为任何子部分,因此 len 为 1。然后它满足 if 条件并尝试对不存在的位置 2 执行某些操作。

在现有代码中,len不是3,所以代码不会进入if block ,因此不会抛出异常。

关于java - 在 MapReduce 中使用 Custom Partitioner 时清空 reduce 输出,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/44910461/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com