分区程序无法正常工作
Partitioner is not working correctly
我正在尝试编写一个 MapReduce 场景的代码,在该场景中我以 JSON 的形式创建了一些用户点击流数据。之后,我编写了 Mapper class 从文件中获取所需的数据,我的 mapper 代码是:-
private final static String URL = "u";
private final static String Country_Code = "c";
private final static String Known_User = "nk";
private final static String Session_Start_time = "hc";
private final static String User_Id = "user";
private final static String Event_Id = "event";
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String aJSONRecord = value.toString();
try {
JSONObject aJSONObject = new JSONObject(aJSONRecord);
StringBuilder aOutputString = new StringBuilder();
aOutputString.append(aJSONObject.get(User_Id).toString()+",");
aOutputString.append(aJSONObject.get(Event_Id).toString()+",");
aOutputString.append(aJSONObject.get(URL).toString()+",");
aOutputString.append(aJSONObject.get(Known_User)+",");
aOutputString.append(aJSONObject.get(Session_Start_time)+",");
aOutputString.append(aJSONObject.get(Country_Code)+",");
context.write(new Text(aOutputString.toString()), key);
System.out.println(aOutputString.toString());
} catch (JSONException e) {
e.printStackTrace();
}
}
}
我的 reducer 代码是:-
public void reduce(Text key, Iterable<LongWritable> values,
Context context) throws IOException, InterruptedException {
String aString = key.toString();
context.write(new Text(aString.trim()), new Text(""));
}
我的分区程序代码是:-
public int getPartition(Text key, LongWritable value, int numPartitions) {
String aRecord = key.toString();
if(aRecord.contains(Country_code_Us)){
return 0;
}else{
return 1;
}
}
这是我的驱动代码
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Click Stream Analyzer");
job.setNumReduceTasks(2);
job.setJarByClass(ClickStreamDriver.class);
job.setMapperClass(ClickStreamMapper.class);
job.setReducerClass(ClickStreamReducer.class);
job.setPartitionerClass(ClickStreamPartitioner.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
这里我试图根据国家代码对我的数据进行分区。但它不起作用,它正在发送单个 reducer 文件中的每条记录,我认为文件不是为 US reduce 创建的文件。
还有一件事,当我看到映射器的输出时,它显示在每条记录的末尾添加了一些额外的 space。
如果我在这里有任何错误,请提出建议。
您的分区问题是由于减速器的数量。如果它是 1,您的所有数据都将发送给它,独立于您 return 从您的分区程序。因此,将 mapred.reduce.tasks
设置为 2 将解决此问题。或者你可以简单地写:
job.setNumReduceTasks(2);
为了有2个你想要的reducer。
除非您有非常特殊的要求,否则您可以为作业参数设置减速器。
mapred.reduce.tasks (in 1.x) & mapreduce.job.reduces(2.x)
或
job.setNumReduceTasks(2)
根据 mark91 的回答。
但使用以下 API 将工作留给 Hadoop fraemork。框架将根据文件和块大小决定 reducer 的数量。
job.setPartitionerClass(HashPartitioner.class);
我已经使用了 NullWritable 并且它有效。现在我可以看到记录被分区到不同的文件中。由于我使用 longwritable 作为 null 值而不是 null writable ,因此在每行的最后添加 space 并且由于这个 US 被列为 "US " 并且分区无法划分订单。
我正在尝试编写一个 MapReduce 场景的代码,在该场景中我以 JSON 的形式创建了一些用户点击流数据。之后,我编写了 Mapper class 从文件中获取所需的数据,我的 mapper 代码是:-
private final static String URL = "u";
private final static String Country_Code = "c";
private final static String Known_User = "nk";
private final static String Session_Start_time = "hc";
private final static String User_Id = "user";
private final static String Event_Id = "event";
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String aJSONRecord = value.toString();
try {
JSONObject aJSONObject = new JSONObject(aJSONRecord);
StringBuilder aOutputString = new StringBuilder();
aOutputString.append(aJSONObject.get(User_Id).toString()+",");
aOutputString.append(aJSONObject.get(Event_Id).toString()+",");
aOutputString.append(aJSONObject.get(URL).toString()+",");
aOutputString.append(aJSONObject.get(Known_User)+",");
aOutputString.append(aJSONObject.get(Session_Start_time)+",");
aOutputString.append(aJSONObject.get(Country_Code)+",");
context.write(new Text(aOutputString.toString()), key);
System.out.println(aOutputString.toString());
} catch (JSONException e) {
e.printStackTrace();
}
}
}
我的 reducer 代码是:-
public void reduce(Text key, Iterable<LongWritable> values,
Context context) throws IOException, InterruptedException {
String aString = key.toString();
context.write(new Text(aString.trim()), new Text(""));
}
我的分区程序代码是:-
public int getPartition(Text key, LongWritable value, int numPartitions) {
String aRecord = key.toString();
if(aRecord.contains(Country_code_Us)){
return 0;
}else{
return 1;
}
}
这是我的驱动代码
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Click Stream Analyzer");
job.setNumReduceTasks(2);
job.setJarByClass(ClickStreamDriver.class);
job.setMapperClass(ClickStreamMapper.class);
job.setReducerClass(ClickStreamReducer.class);
job.setPartitionerClass(ClickStreamPartitioner.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
这里我试图根据国家代码对我的数据进行分区。但它不起作用,它正在发送单个 reducer 文件中的每条记录,我认为文件不是为 US reduce 创建的文件。
还有一件事,当我看到映射器的输出时,它显示在每条记录的末尾添加了一些额外的 space。
如果我在这里有任何错误,请提出建议。
您的分区问题是由于减速器的数量。如果它是 1,您的所有数据都将发送给它,独立于您 return 从您的分区程序。因此,将 mapred.reduce.tasks
设置为 2 将解决此问题。或者你可以简单地写:
job.setNumReduceTasks(2);
为了有2个你想要的reducer。
除非您有非常特殊的要求,否则您可以为作业参数设置减速器。
mapred.reduce.tasks (in 1.x) & mapreduce.job.reduces(2.x)
或
job.setNumReduceTasks(2)
根据 mark91 的回答。
但使用以下 API 将工作留给 Hadoop fraemork。框架将根据文件和块大小决定 reducer 的数量。
job.setPartitionerClass(HashPartitioner.class);
我已经使用了 NullWritable 并且它有效。现在我可以看到记录被分区到不同的文件中。由于我使用 longwritable 作为 null 值而不是 null writable ,因此在每行的最后添加 space 并且由于这个 US 被列为 "US " 并且分区无法划分订单。