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;
}
}
Util.java 文件内容
/*
@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;
}
}
TagBean.java 文件内容
/*
@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("\t");
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));
}
}
}
}
TaggenMapper.java 文件内容
/*
@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));
}
}
TaggenReducer.java 文件内容
/*
@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));
}
}
TaggenMapper2.java 文件内容
/*
@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);
}
}
TaggenReducer2.java 文件内容
/*
@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);
}
}
}