java - 多个单词的 Hadoop WordCount 未获取公共变量
问题描述
我有一个简单的 Hadoop 程序,我需要为 mu 大学的一篇论文实施该程序。这是一个替代的 WordCount 问题,它应该使组合的 Text() 字符串具有 n 个单词,并且仅与 reducer 总结那些 >= 大于 k 出现的字符串。我已将 n 和 k 整数放在输入和输出文件夹(args[3] 和 args[4])之后从命令行捕获。问题是 n 和 k 在 mapper 和 reducer 中使用时是空的,尽管从命令中正确获取了它们的值。代码如下,有什么问题?
public class MultiWordCount {
public static int n;
public static int k;
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
private StringBuilder phrase = new StringBuilder();
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
for (int i=0; i<n; i++) {
if (itr.hasMoreTokens()) {
phrase.append(itr.nextToken());
phrase.append(" ");
}
}
word.set(phrase.toString());
context.write(word, one);
phrase.setLength(0);
}
}
}
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
if(sum >= k) {
result.set(sum);
context.write(key, result);
}
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
n = Integer.parseInt(args[2]);
k = Integer.parseInt(args[3]);
Job job = Job.getInstance(conf, "multi word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
解决方案
尽管您在此处基于 Java 的逻辑看起来很合理,但在 Hadoop 中实现的 Map 和 Reduce 函数比人们想象的更加短视或独立。更准确地说,您在父类中声明公共静态变量并在驱动程序/主函数中对其进行初始化,但映射器/归约器实例对驱动程序没有任何访问权限,而只能访问它们在TokenizerMapper
和IntSumReducer
类中的严格范围。这就是当您查看映射器n
和k
减速器内部时看起来很空的原因。
由于您的程序只有一个作业并且在单个 Hadoop 中执行,因此此处Configuration
不需要Hadoop 计数器。您可以在执行 Map 和 Reduce 函数之前,通过和类中的函数声明Configuration
将由每个映射器和化简器访问的基于值。setup
TokenizerMapper
IntSumReducer
要声明这些类型的值以便将它们传递给 MapReduce 函数,您可以在 driver/main 方法中执行类似的操作:
conf.set("n", args[2]);
然后在and的方法中访问这个值(同时将其转换String
为int
):setup
TokenizerMapper
IntSumReducer
n = Integer.parseInt(context.getConfiguration().get("n"));
所以程序可以如下所示:
import org.apache.hadoop.conf.Configuration;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.Counters;
import java.io.*;
import java.io.IOException;
import java.util.*;
import java.nio.charset.StandardCharsets;
public class MultiWordCount
{
public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>
{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
private StringBuilder phrase = new StringBuilder();
private int n;
protected void setup(Context context) throws IOException, InterruptedException
{
n = Integer.parseInt(context.getConfiguration().get("n"));
}
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens())
{
for (int i = 0; i < n; i++)
{
if (itr.hasMoreTokens())
{
phrase.append(itr.nextToken());
phrase.append(" ");
}
}
word.set(phrase.toString());
context.write(word, one);
phrase.setLength(0);
}
}
}
public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable>
{
private IntWritable result = new IntWritable();
private int k;
protected void setup(Context context) throws IOException, InterruptedException
{
k = Integer.parseInt(context.getConfiguration().get("k"));
}
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
{
int sum = 0;
for (IntWritable val : values)
sum += val.get();
if(sum >= k)
{
result.set(sum);
context.write(key, result);
}
}
}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
conf.set("n", args[2]);
conf.set("k", args[3]);
FileSystem fs = FileSystem.get(conf);
if(fs.exists(new Path(args[1])))
fs.delete(new Path(args[1]), true);
Job job = Job.getInstance(conf, "Multi Word Count");
job.setJarByClass(MultiWordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
对于n=3
and k=1
,输出看起来像这样(使用一些带有 Franz Kafka 句子的文本文件,如此处所示):
推荐阅读
- python - pyside2 QT 我在填充我的 QListwidget 的图标标签周围有一条虚线,我想摆脱它
- angularjs - ng-required 将复选框从 false 更改为 undefined
- google-apps-script - 谷歌环聊聊天机器人可以消息机器人吗?
- printf - 为什么超过 100 的值变为负数?
- angular - Firefox 上的 Angular Material mat-checkbox 问题
- reactjs - 使用 geofirestore 添加、设置、更新文档
- javascript - 从服务器,如何识别 javascript fetch() 调用(asp.net mvc)
- android - 在drawable中添加和定位drawable
- python - 计算 django 电子商务中的订单总额
- class - 在另一个类 C++ 的方法中创建类对象