Hadoop基础--统计商家id的标签数案例分析

　　　　　　　　　　Hadoop基础--统计商家id的标签数案例分析

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正杰

版权声明：原创作品，谢绝转载！否则将追究法律责任。

一.项目需求

 　　将“temptags.txt”中的数据进行分析，统计出商家id的评论标签数量，由于博客园无法上传大文件的文本，因此我把该文本的内容放在博客园的另一个链接了（需要的戳我），如果网页打不开的话也就可以去百度云盘里下载副本，链接：https://pan.baidu.com/s/1daRiwOVe6ohn42fTv6ysJg 密码：h6er。

　　实现效果如下：

二.代码实现

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import java.util.ArrayList;
import java.util.List;

public class Util {
    public static List<String> taggen(String comment){
        //解析传进来的字符串
        JSONObject jo = JSON.parseObject(comment);
        //拿到包含商家评论的相关的标签
        JSONArray jArray = jo.getJSONArray("extInfoList");
        //过滤掉不含商家评论的标签
        if(jArray != null && jArray.size() != 0){
            //定义一个空的有序集合
            List<String> list = new ArrayList<String>();
            //通过jArray得到第一个json串，作为json对象
            JSONObject jo2 = jArray.getJSONObject(0);
            //进一步拿到商家评论的相关的标签
            JSONArray jArray2 = jo2.getJSONArray("values");
            //进一步过滤掉不含商家评论的标签
            if(jArray2 != null && jArray2.size() != 0){
                for (Object obj : jArray2) {
                    //将商检评论的标签添加到我们定义的集合中
                    list.add(obj.toString());
                }
                return list;
            }
        }
        return null;
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class TagBean implements WritableComparable<TagBean> {
    private String tag;
    private int count;

    /**
     *         compareTo() 方法用于将 Number 对象与方法的参数进行比较。可用于比较 Byte, Long, Integer等。
     * 该方法用于两个相同数据类型的比较，两个不同类型的数据不能用此方法来比较。
     */
    public int compareTo(TagBean o) {
       if(o.count == this.count){
           return this.getTag().compareTo(o.getTag());
       }
       return o.count - this.count;
    }

    public void write(DataOutput out) throws IOException {
        out.writeUTF(tag);
        out.writeInt(count);
    }
    public void readFields(DataInput in) throws IOException {
        tag = in.readUTF();
        count = in.readInt();
    }
    public String getTag() {
        return tag;
    }
    public void setTag(String tag) {
        this.tag = tag;
    }
    public int getCount() {
        return count;
    }
    public void setCount(int count) {
        this.count = count;
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.List;

public class TaggenMapper extends Mapper<LongWritable,Text,Text,IntWritable> {

    /**
     *
     * @param key                   //这是读取行的偏移量
     * @param value                 //这是这是的数据，每条数据格式都类似，比如：70611801    {"reviewPics":null,"extInfoList":null,"expenseList":null,"reviewIndexes":[1,2],"scoreList":[{"score":4,"title":"环境","desc":""},{"score":5,"title":"服务","desc":""},{"score":4,"title":"口味","desc":""}]}
     * @param context
     */
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] arr = line.split("	");
        String id = arr[0];
        String json = arr[1];
        //我们对数据进行解析，把用户评论的标签数都整合起来，用tags变量来接受数据
        List<String> tags = Util.taggen(json);
        //如果tags没数据，则不写入
        if(tags != null && tags.size() != 0){
            for(String tag : tags){
                //给数据打标签，最终结果类似于 : 70611801_价格实惠
                String compKey = id+ "_"+ tag;
                context.write(new Text(compKey), new IntWritable(1));
            }
        }
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class TaggenReducer  extends Reducer<Text, IntWritable , Text, IntWritable> {
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        Integer sum = 0;
        for(IntWritable value : values){
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class TaggenMapper2 extends Mapper<Text,Text, Text,Text> {


    //89223651_价格实惠     8
    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {

        //此时的key其实就是上一个MapReduce的结果，比如：70611801_价格实惠
        String compkey = key.toString();
        //我们将结果进行拆分，取出来id，比如 : 70611801
        String id = compkey.split("_")[0];
        //我们将结果进行拆分，取出来tag，比如 : 价格实惠
        String tag = compkey.split("_")[1];
        //此时的value的值其实就是对key的一个计数 : 比如key出现的次数为8
        String sum = value.toString();
        //这个时候我们就是将tag和之前的value进行拼接，得出结果如下 : 价格实惠_8
        String newVal = tag + "_" + sum;
        //最后将我们重写组合的key和value重新分发给reduce
        context.write(new Text(id), new Text(newVal));
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.TreeSet;

public class TaggenReducer2 extends Reducer<Text,Text,Text,Text> {

    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //使用TreeSet的目的是实现去重并排序
        TreeSet<TagBean> ts = new TreeSet<TagBean>();
        //迭代value，并将其放入treeSet，而TreeSet使用的排序是自定义(TagBean)的。
        for(Text value:values){
            String[] arr = value.toString().split("_");
            String tag = arr[0];
            int count = Integer.parseInt(arr[1]);
            TagBean tagBean = new TagBean();
            tagBean.setTag(tag);
            tagBean.setCount(count);
            ts.add(tagBean);
        }
        //迭代TreeSet中的TagBean，并得到tag和count，放进StringBuffer
        StringBuffer sb = new StringBuffer();
        for (TagBean tb : ts) {
            String tag = tb.getTag();
            int count = tb.getCount();
            String val = tag+"_"+count;
            sb.append(val+ ",");
        }
        String newVal = sb.toString().substring(0, sb.length() -1);
        //经过上面的整理，可以把同一个商家id的所有标签整个到一起，最终发送出去FileOutputFormat,最终格式为类型如 : 83084036 价格实惠_1,干净卫生_1
        context.write(key,new Text(newVal));
        //TreeSet置空
        ts.clear();
        //StringBuffer置空
        sb.setLength(0);
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.taggen;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class TaggenApp {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        FileSystem fs = FileSystem.get(conf);
        Job job = Job.getInstance(conf);
        job.setJobName("taggen");
        job.setJarByClass(TaggenApp.class);
        job.setMapperClass(TaggenMapper.class);
        job.setReducerClass(TaggenReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        Path outPath = new Path("D:\10.Java\IDE\yhinzhengjieData\MyHadoop\out");
        if(fs.exists(outPath)){
            fs.delete(outPath,true);
        }
        FileInputFormat.addInputPath(job,new Path("D:\10.Java\IDE\yhinzhengjieData\MyHadoop\temptags.txt"));
        FileOutputFormat.setOutputPath(job,outPath);
        if(job.waitForCompletion(true)){
            Job job2 = Job.getInstance(conf);
            job2.setJobName("taggen2");
            job2.setJarByClass(TaggenApp.class);
            job2.setMapperClass(TaggenMapper2.class);
            job2.setReducerClass(TaggenReducer2.class);
            job2.setOutputKeyClass(Text.class);
            job2.setOutputValueClass(Text.class);
            job2.setInputFormatClass(KeyValueTextInputFormat.class);
            Path outPath2 = new Path("D:\10.Java\IDE\yhinzhengjieData\MyHadoop\out2");
            if(fs.exists(outPath2)){
                fs.delete(outPath2,true);
            }
            FileInputFormat.addInputPath(job2,outPath);
            FileOutputFormat.setOutputPath(job2,outPath2);
            job2.waitForCompletion(true);
        }
    }
}