java - hadoop 上的 JSON 处理

Question

我正在尝试对 json 文件运行 map reduce。输入文件的格式如下。

{"Id":1, "title":"A list of SaaS management resources to help kickstart and augment your efforts","category":"business"}
{"Id":2, "title":"All Over the Board: 1Working on a  23 (Temp) Dream","category":"business"}
{"Id":3, "title":"Tulsa Web Design","category":"business"}

我的 reduce 函数的预期输出如下。

1 business A 1
1 business list 1
1 business of 1

下面是我用来读取 json 文件的代码，获取所需的值，然后将其转换为字符串。通过将字符串拆分为键值来对该字符串进行字数统计，其中键为 Id + 类别 + 标题中的每个词，值为 1。

import java.io.IOException;
import java.util.StringTokenizer;

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.json.*;

public class mr1 {

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{

        private final static IntWritable ONE = new IntWritable(1);
        private Text word = new Text();


        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

               String title;
                       String Id;
                       String category;
                       String line;
                       String valueLine;

            try {

                    line = value.toString();


                    JSONObject obj = new JSONObject(parser.parse(line));

                    title = (String) obj.get("title");
                    category = (String) obj.get("category");
                    Id = (String) obj.get("Id");

                    title = title.replaceAll("[!?,:()1-9]", "");
                    String[] strs = title.split("\\s+");

                    StringBuilder sb = new StringBuilder();

                            for(int i=0; i < strs.length; i++) {
                                sb.append(strs[i]+" ");
                            }

                    // valueLine  = 1 business Tulsa Web Design

                    valueLine = Id + " " + category + " " + sb.toString();

                    StringTokenizer itr = new StringTokenizer(valueLine);
                    String IndexAndCategory = "";

                    IndexAndCategory += itr.nextToken() + " ";
                    IndexAndCategory += itr.nextToken() + " ";


                    while(itr.hasMoreTokens()) {

                        word.set(IndexAndCategory + itr.nextToken());
                        context.write(word, ONE);
                    }

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

    }

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

        private IntWritable result = new IntWritable();

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

                int sum = 0;
                for(IntWritable val : values) {     
                    sum += val.get();               
                }

                result.set(sum);
                context.write(key, result);
        }   
    }


    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "mr1");
        job.setJarByClass(mr1.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
                job.setReducerClass(IntSumReducer.class);
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(IntWritable.class);
                job.setNumReduceTasks(1);
                FileInputFormat.addInputPath(job, new Path(args[0]));
                FileOutputFormat.setOutputPath(job, new Path(args[1]));
                System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}

我能够编译代码创建 jar 文件，当我在 hadoop 上运行它时出现以下错误。

Error: java.lang.ClassNotFoundException: org.json.JSONException
        at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
        at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
        at java.security.AccessController.doPrivileged(Native Method)
        at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
        at java.lang.Class.forName0(Native Method)
        at java.lang.Class.forName(Class.java:270)
        at org.apache.hadoop.conf.Configuration.getClassByNameOrNull(Configuration.java:1986)
        at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:1951)
        at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2045)
        at org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:196)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:742)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:415)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1614)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:163)

我用于此代码的 jar 文件如下；

1. hadoop-common-2.5.0.jar 
2. json-20160212.jar
3. hadoop-mapreduce-client-core-2.5.0.jar

我想问题出在我将行转换为 JSONObject 的地方，但我不确定这个问题。感谢任何解决此问题的帮助。

Answer 1

请检查您是否忘记在 Hadoop 作业 jar 中添加 JSON jar。这可能对您有帮助:http://tikalk.com/build-your-first-hadoop-project-maven