你好,游客 登录 注册 搜索
背景:
阅读新闻

Hadoop(五)——核心编程MapReduce(下)

[日期:2015-09-14] 来源:CSDN博客  作者: [字体: ]

上篇博客最后我们讲述了 Word Count  hadoop 官方源码,主要看 map 类的编写规则,入参(从文件)出参(经过shuffle, com biner 过程给 reduce ), reduce 的编写规则,入参(从 map 类中获取),出参(想要的结果输出到文件中)。下边我们再进一步通过几个例子(在 hadoop 实战中摘取),来加深 map-reduce 的编程规则,至于具体到 map  reduce 内部,如何处理数据,则涉及到算法,因情况而异。在这里我们学到 mapreduce 的执行流程,编写规则等即可。

一,数据去重:顾名思义就是计算数据中的类别,将数据多余一次只显示一次。很简单的一个过程,对于 map reduce , 直接看代码,注意里边的注释 :

package job;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.apache.hadoop.util.GenericOptionsParser;

/**
 * 一,数据去重的Map-Reduce实例
 * 
 * @author ljh
 * 
 */
public class Dedup {

/*
 * 1,Map将输入中的value复制到输出数据的key上,并直接输出,value值无所谓,利用shuffle这个工程进行key相同汇总
 */
public static class Map extends Mapper<Object, Text, Text, Text> {
// 定义line存储每行的数据
private static Text line = new Text();

// map函数直接将value复制给line,然后输出即可
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
line = value;
context.write(line, new Text(""));
}
}

/*
 * 2,reduce将输入的key复制到输出数据的key上,并直接输出
 */
public static class Reduce extends Reducer<Text, Text, Text, Text> {
// reduce函数,利用shuffle处理好的,直接输出即可,比较简单
public void reduce(Text key, Text values, Context context)
throws IOException, InterruptedException {
context.write(key, new Text(""));
}
}

/*
 * 3,main方法
 */
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
// 获取输入文件和输出文件的地址
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage:  in and out");
System.exit(2);
}

Job job = new Job(conf, "Data deduplication");
job.setJarByClass(Dedup.class);
job.setMapperClass(Map.class);
job.setCombinerClass(Reduce.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);

}
}

二,排序:这个在我们写 SQL 中利用 order by 列名 desc/asc 进行排序即可, Java中我们有基本的冒泡排序,选择排序,顺序排序等,还有堆排序,归并排序,基数排序等,这里看两篇文章: http://blog.csdn.net/ygc87/article/details/7208082 http://www.cnblogs.com/liuling/p/2013-7-24-01.html    。而在 map reduce 中就有默认的排序,如果 key 封装的 int  IntWritable, 则会按照数字的大小进行排序;如果封装的 String  Text, 则会按照字典的顺序字符串进行排序。好,看下边的程序代码:

package job;

import java.io.IOException;

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.apache.hadoop.util.GenericOptionsParser;

/**
 * 二,排序的Map-Reduce实例:排序为一些数字
 * @author ljh
 *
 */
public class Sort {
/*
 * 1,map将输入中的value化成IntWritable类型,作为输出的key
 */
public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{
//用来存储数据,由于是数字的排序,直接使用IntWritable即可
private static IntWritable data=new IntWritable();

public void map(Object key,Text value,Context context)throws IOException,InterruptedException{
String line=value.toString();
data.set(Integer.parseInt(line));
context.write(data, new IntWritable(1));
}
}

public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
//第一行的行号设置为1
private static IntWritable linenum=new IntWritable(1);

//已经自动排序好了,输出即可,这里我们来设置一下行号自动加1即可
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException,InterruptedException{
for(IntWritable val:values){
context.write(linenum, key);
linenum=new IntWritable(linenum.get()+1);
}	
}
}

/*
 * 3,main方法
 */
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
// 获取输入文件和输出文件的地址
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage:  in and out");
System.exit(2);
}

Job job = new Job(conf, "Data Sort");
job.setJarByClass(Sort.class);
job.setMapperClass(Map.class);
//这里没有设置CombinerClass,CombinerClass是为了简述网络流量,为了是输出的数据只是必要的,例如上边的去重,先在本地进行去重,再进行传输整合,而这个例子进行的排序是没有必要的。
//job.setCombinerClass(Reduce.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);

}
}

三,单表关联:也就是我们经常见到的一张表中的父子关系,有其树形结构。在orcle 中我们有经典的 sql  select * from tb_menu m start with m.id=1   connect by m.parent=prior m.id;

 Java 中我们使用递归操作来进行操作。下边通过 child -parent 表,来寻找grandchild-grandparent 关系表,也就是通过父子关系数据,找出爷孙关系的数据对应,好,看下 Map -reduce 中操作:

package job;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.apache.hadoop.util.GenericOptionsParser;

/**
 * 三,单表关联查询父子关系的Map-Reduce实例:父子关系的表数据
 * 
 * @author ljh
 * 
 */
public class STJoin {
// 判断是否是首行,因为首行是标题:输入文件为child-parent,输出文件为grandchild-grandparent
public static int time = 0;

/*
 * 1,map将输入分割为child和parent,然后正序输出依次作为右表,反序依次输出作为左表,需要注意在value中加上左表和右表的标识
 */
public static class Map extends Mapper<Object, Text, Text, Text> {

public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String childName = new String();// 孩子的名称存储
String parentName = new String();// 父亲的名称存储
String relationType = new String();// 左表右表的标识1,表示左表,2表示右表
String line = value.toString();// 输入的每一行转化为字符串

int i = 0;
//当检索line的第i个字符不为tab制表符使,进行++,也就是为了下边的切割
while (line.charAt(i) != '	') {
i++;
}

//将child和parent放到数组当中,利用上边找到的分割符
String[] values = { line.substring(0, i), line.substring(i + 1) };

//当不是第一行标题的时候
if (values[0].compareTo("child") != 0) {
//获取child和parent
childName = values[0];
parentName = values[1];
// 左表,输出,左右表的key是一对父子关系
relationType = "1";
context.write(new Text(values[1]), new Text(relationType + "+"
+ childName + "+" + parentName));

// 右表,输出
relationType = "2";
context.write(new Text(values[0]), new Text(relationType + "+"
+ childName + "+" + parentName));
}
}
}

/*
 * 2,reduce:拿到map输出的左右表是一对负责关系表
 * 左表:{tom,[{1,lucy,tom},{1,lili,tom}]
 * 右表:{tom,[{2,tom,hali},{2,tom,karry}]
 * 其实应该是合并好的:{tom,[{1,lucy,tom},{1,lili,tom},{2,tom,hali},{2,tom,karry}]
 * 所有,我们取tom的父亲,和tom的孩子,就得到了爷孙关系
 */
public static class Reduce extends Reducer<Text, Text, Text, Text> {

public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
//如果是第一行,则输出表头信息
if (time == 0) {
context.write(new Text("grandchild"), new Text("grandparent"));
time++;
}

int grandchildnum = 0;//表示孙子的个数
String grandchild[] = new String[10];//存放孙子的数组
int grandparentnum = 0;//表示爷爷的个数
String grandparent[] = new String[10];//存放爷爷的数组
Iterator ite = values.iterator();//每一个儿子对应的父亲列表

//如果有父亲
while (ite.hasNext()) {
//取出父亲的集合
String record = ite.next().toString();
//父亲集合的长度
int len = record.length();
int i = 2;
if (len == 0) continue;//结束本次循环

char relationType = record.charAt(0);//取出是父表还是子表
String childname = new String();
String parentname = new String();
//获取value-list中的value的child
while (record.charAt(i) != '+') {
childname = childname + record.charAt(i);
i++;
}
i = i + 1;

// 获取value-list中的value的parent
while (i < len) {
parentname = parentname + record.charAt(i);
i++;
}
//如果是父表,取出孩子,如果是子表,取出父亲,组成了爷孙关系
if (relationType == '1') {
grandchild[grandchildnum] = childname;
grandchildnum++;
} else {
grandparent[grandparentnum] = parentname;
grandparentnum++;
}
}
//爷孙求笛卡尔积,看有多少对组合
if (grandparentnum != 0 && grandchildnum != 0) {
for (int m = 0; m < grandchildnum; m++) {
for (int n = 0; n < grandparentnum; n++) {
context.write(new Text(grandchild[m]), new Text(
grandparent[n]));
}
}
}
}
}

/*
 * 3,main方法
 */
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
// 获取输入文件和输出文件的地址
String[] otherArgs = new GenericOptionsParser(conf, args)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage:  in and out");
System.exit(2);
}

Job job = new Job(conf, "single table join");
job.setJarByClass(STJoin.class);
job.setMapperClass(Map.class);
// job.setCombinerClass(Reduce.class);
job.setReducerClass(Reduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);

}
}

 

四,多表关联:其实上边的单表关联,我们就是将其转化为两张表进行关联的,其中的关键就是将两表中公用的字段设置为 key ,我们就能得到关于这个 key 的集合,其中有左表的数据,有右表的数据,然后我们取出来进行笛卡尔积即可。这里不再给出代码,大家好好思考一下,其实多表关联,比单表关联还简单,因为多表不需要我们进行抽象了,单表还需要我们抽象出两张表。

好,这篇简单讲述了 map -reduce 的几个例子,虽然简单,但是确实不得不走的路程,慢慢熟悉编写 map-reduce 的基本知识,然后通过场景,通过各种算法,通过灵活的大脑,来写出属于实际场景有价值的 map-reduce 来。





收藏 推荐 打印 | 录入:elainebo | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款