分区程序工作不正常

ie3xauqp  于 2021-06-02  发布在  Hadoop
关注(0)|答案(3)|浏览(524)

我正在尝试编写一个mapreduce场景,其中我以json的形式创建了一些用户clickstream数据。之后,我编写了mapper类来从文件中获取所需的数据,我的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();
    }
}

}
我的代码是:-

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文件发送每一个记录,我想其他文件,然后为我们创建reduce。
还有一件事,当我看到Map器的输出时,它会在每条记录的末尾显示一些额外的空间。
如果我在这里犯了什么错误,请提出建议。

e7arh2l6

e7arh2l61#

我使用了nullwriteable,它可以工作。现在我可以看到记录在不同的文件中被分区。因为我使用longwritable作为null值而不是null writable,所以在每行的最后添加了空格,因此us被列为“us”,分区无法划分顺序。

yk9xbfzb

yk9xbfzb2#

分区的问题是由于减速器的数量。如果它是1,所有的数据都将被发送到它,独立地从分区器返回。因此,设置 mapred.reduce.tasks 到2将解决这个问题。或者你可以简单地写下:

job.setNumReduceTasks(2);

为了有两个你想要的减速器。

mrwjdhj3

mrwjdhj33#

除非您有非常具体的要求,您可以设置减速器如下工作参数。

mapred.reduce.tasks (in 1.x) & mapreduce.job.reduces(2.x)

或者 job.setNumReduceTasks(2) 根据mark91答案。
但是把工作留给hadoopfraemork使用belowapi。框架将根据文件和块的大小来决定缩减器的数量。

job.setPartitionerClass(HashPartitioner.class);

相关问题