我在编写 mapreduce 函数时遇到了一些问题。 我想解决以下问题:

我有一个带有 1mio JSONObject 的 JSON 文件，如下所示:

 {"_id":3951,"title":"Two Family House (2000)","genres":["Drama"],"ratings":[{"userId":173,"rating":5},{"userId":195,"rating":5},{"userId":411,"rating":4},{"userId":593,"rating":2},{"userId":629,"rating":3},{"userId":830,"rating":3},{"userId":838,"rating":5},{"userId":850,"rating":4},{"userId":856,"rating":4},{"userId":862,"rating":5},{"userId":889,"rating":1},{"userId":928,"rating":5},{"userId":986,"rating":4},{"userId":1001,"rating":5},{"userId":1069,"rating":3},{"userId":1168,"rating":3},{"userId":1173,"rating":2},{"userId":1242,"rating":3},{"userId":1266,"rating":5},{"userId":1331,"rating":5},{"userId":1417,"rating":5},{"userId":1470,"rating":4},{"userId":1474,"rating":5},{"userId":1615,"rating":3},{"userId":1625,"rating":4},{"userId":1733,"rating":4},{"userId":1799,"rating":4},{"userId":1865,"rating":5},{"userId":1877,"rating":5},{"userId":1897,"rating":5},{"userId":1946,"rating":4},{"userId":2031,"rating":4},{"userId":2129,"rating":2},{"userId":2353,"rating":4},{"userId":2986,"rating":4},{"userId":3940,"rating":4},{"userId":3985,"rating":3},{"userId":4025,"rating":5},{"userId":4727,"rating":3},{"userId":5333,"rating":3}]}

还有更多....

一个 JSON 对象是一部电影，其中包含一个数组评级。我想计算 JSON 文件中的所有评分。

我在 IntelliJ 中创建了一个 Maven Proct，它依赖于 Hadoop 和 JSON Parser。我的 MapReduce 类是这样的:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
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.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import java.io.IOException;
import java.util.Iterator;

public class RatingCounter {

public static class RatingMapper extends Mapper<JSONObject, Text, Text, Text>{

    private Text id = new Text();
    private Text ratingAnzahl = new Text();

    public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException{
        JSONParser parser = new JSONParser();

        try {
            Object obj = parser.parse(value.toString());
            JSONObject jsonObject = (JSONObject) obj;

            String movieId = (String) jsonObject.get("_id");

            int count = 0;
            // loop array
            JSONArray ratings = (JSONArray) jsonObject.get("ratings");
            Iterator<String> iterator = ratings.iterator();
            while (iterator.hasNext()) {
                count++;
            }

        } catch (ParseException e) {
            e.printStackTrace();
        }
    }
}


public static class RatingReducer extends Reducer<Text, Text, Text, Text> {

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        Text resultValue = new Text();

        int allRatings = 0;

        while (values.hasNext()){
            allRatings += Integer.parseInt(values.toString());

        }
        resultValue.set(""+allRatings);
        context.write(key, resultValue);
    }
}

public static void main (String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = new Job(conf, "ratings count");
    job.setJarByClass(RatingCounter.class);
    job.setMapperClass(RatingMapper.class);
    job.setReducerClass(RatingReducer.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);
}
}

我不知道如何在 Mapper 和 Reducer 中编写函数。有人可以帮我吗？

最佳答案

我对您的映射器和缩减器做了一些更改。

首先，对于您的映射器，您没有在任何地方编写输出，并且您在扩展 Mapper 类时的语法也是错误的(有争议的)。任何映射器的第一个输入都是 LongWritable(或 Object 类型)行偏移量。您可以注意到以下变化

public static class RatingMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

    public void map(LongWritable key, Text value, Context context) throws IOException, ParseException{
        JSONParser parser = new JSONParser();

        Object obj = parser.parse(value.toString());
        JSONObject jsonObject = (JSONObject) obj;

        String movieId = (String) jsonObject.get("_id");

        JSONArray ratings = (JSONArray) jsonObject.get("ratings");

        context.write(new Text(movieId), new IntWritable(ratings.size()) );
    }
}

注意这里，map的输出是用context.write

现在，进入你的 Reducer 一些事情会因为我在映射器中所做的更改而改变。此外，由于您的 Number of Ratings 将始终是一个整数，因此您无需将其转换为 Text，使用 parseInt 然后转换再次 Text。

public static class RatingReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        int allRatings = 0;

        while (values.hasNext()){
            allRatings += value.get();
        }
        context.write(key, new IntWritable(resultValue));
    }
}

关于json - 带有 JSON 文件和 JSONParser 的 MapReduce 函数，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/44398273/