这篇文章将为大家详细讲解有关Hadoop中MapReduce常用算法有哪些，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。
hadoop fs -mkdir /import
创建一个或者多个文本，上传
hadoop fs -put test.txt /import/
package com.cuiweiyou.sort;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;
//hadoop默认排序：
//如果k2、v2类型是Text-文本，结果是按照字典顺序
//如果k2、v2类型是LongWritable-数字，结果是按照数字大小顺序
public class SortTest {
/**
* 内部类：映射器 Mapper<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyMapper extends Mapper<LongWritable, Text, LongWritable, NullWritable> {
/**
* 重写map方法
*/
public void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
//这里v1转为k2-数字类型，舍弃k1。null为v2
context.write(new LongWritable(Long.parseLong(v1.toString())), NullWritable.get());
//因为v1可能重复，这时，k2也是可能有重复的
}
}
/**
* 内部类：拆分器 Reducer<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyReducer extends Reducer<LongWritable, NullWritable, LongWritable, NullWritable> {
/**
* 重写reduce方法
* 在此方法执行前，有个shuffle过程，会根据k2将对应的v2归并为v2[...]
*/
protected void reduce(LongWritable k2, Iterable<NullWritable> v2, Reducer<LongWritable, Context context) throws IOException, InterruptedException {
//k2=>k3, v2[...]舍弃。null => v3
context.write(k2, NullWritable.get());
//此时，k3如果发生重复，根据默认算法会发生覆盖，即最终仅保存一个k3
}
}
public static void main(String[] args) throws Exception {
// 声明配置信息
Configuration conf = new Configuration();
conf.set("fs.default.name", "hdfs://localhost:9000");
// 创建作业
Job job = new Job(conf, "SortTest");
job.setJarByClass(SortTest.class);
// 设置mr
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
// 设置输出类型，和Context上下文对象write的参数类型一致
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(NullWritable.class);
// 设置输入输出路径
FileInputFormat.setInputPaths(job, new Path("/import/"));
FileOutputFormat.setOutputPath(job, new Path("/out"));
// 执行
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
可以看到，不仅排序而且去重了。
需求：查取手机号有哪些。这里的思路和上面排序算法的思路是一致的，仅仅多了分割出手机号这一步骤。
创建两个文本，手动输入一些测试内容。每个字段用制表符隔开。日期，电话，地址，方式，数据量。
/**
* 映射器 Mapper<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
/**
* 重写map方法
*/
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//按照制表符进行分割
String[] tels = v1.toString().split("\t");
//k1 => k2-第2列手机号，null => v2
context.write(new Text(tels[1]), NullWritable.get());
}
}
/************************************************************
* 在map后，reduce前，有个shuffle过程，会根据k2将对应的v2归并为v2[...]
***********************************************************/
/**
* 拆分器 Reducer<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
/**
* 重写reduce方法
*/
protected void reduce(Text k2, Iterable<NullWritable> v2, Context context) throws IOException ,InterruptedException {
//此时，k3如果发生重复，根据默认算法会发生覆盖，即最终仅保存一个k3，达到去重到效果
context.write(k2, NullWritable.get());
}
}
// 设置输出类型，和Context上下文对象write的参数类型一致 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class);
需求：查询在北京地区发生的上网记录。思路同上，当写出 k2 、 v2 时加一个判断即可。
同上。
/**
* 内部类：映射器 Mapper<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
/**
* 重写map方法
*/
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//按照制表符进行分割
final String[] adds = v1.toString().split("\t");
//地址在第3列
//k1 => k2-地址，null => v2
if(adds[2].equals("beijing")){
context.write(new Text(v1.toString()), NullWritable.get());
}
}
}
/**
* 内部类：拆分器 Reducer<KEY_IN, VALUE_IN, KEY_OUT, VALUE_OUT>
*/
public static class MyReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
/**
* 重写reduce方法
*/
protected void reduce(Text k2, Iterable<NullWritable> v2, Context context) throws IOException ,InterruptedException {
context.write(k2, NullWritable.get());
}
}
// 设置输出类型，和Context上下文对象write的参数类型一致 job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class);
这个算法非常经典，面试必问。实现这个效果的算法也很多。下面是个简单的示例。
需求：找到流量最大值；找出前5个最大值。
同上。
//map
public static class MyMapper extends Mapper<LongWritable, Text, LongWritable, NullWritable> {
//首先创建一个临时变量，保存一个可存储的最小值：Long.MIN_VALUE=-9223372036854775808
long temp = Long.MIN_VALUE;
//找出最大值
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//按照制表符进行分割
final String[] flows = v1.toString().split("\t");
//将文本转数值
final long val = Long.parseLong(flows[4]);
//如果v1比临时变量大，则保存v1的值
if(temp<val){
temp = val;
}
}
/** ---此方法在全部的map任务结束后执行一次。这时仅输出临时变量到最大值--- **/
protected void cleanup(Context context) throws IOException ,InterruptedException {
context.write(new LongWritable(temp), NullWritable.get());
System.out.println("文件读取完毕");
}
}
//reduce
public static class MyReducer extends Reducer<LongWritable, NullWritable, LongWritable, NullWritable> {
//临时变量
Long temp = Long.MIN_VALUE;
//因为一个文件得到一个最大值，再次将这些值比对，得到最大的
protected void reduce(LongWritable k2, Iterable<NullWritable> v2, Context context) throws IOException ,InterruptedException {
long long1 = Long.parseLong(k2.toString());
//如果k2比临时变量大，则保存k2的值
if(temp<long1){
temp = long1;
}
}
/** ！！！此方法在全部的reduce任务结束后执行一次。这时仅输出临时变量到最大值！！！ **/
protected void cleanup(Context context) throws IOException, InterruptedException {
context.write(new LongWritable(temp), NullWritable.get());
}
}
// 设置输出类型 job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(NullWritable.class);
//map
public static class MyMapper extends Mapper<LongWritable, Text, LongWritable, NullWritable> {
//首先创建一个临时变量，保存一个可存储的最小值：Long.MIN_VALUE=-9223372036854775808
long temp = Long.MIN_VALUE;
//Top5存储空间
long[] tops;
/** 次方法在run中调用，在全部map之前执行一次 **/
protected void setup(Context context) {
//初始化数组长度为5
tops = new long[5];
}
//找出最大值
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//按照制表符进行分割
final String[] flows = v1.toString().split("\t");
//将文本转数值
final long val = Long.parseLong(flows[4]);
//保存在0索引
tops[0] = val;
//排序后最大值在最后一个索引，这样从后到前依次减小
Arrays.sort(tops);
}
/** ---此方法在全部到map任务结束后执行一次。这时仅输出临时变量到最大值--- **/
protected void cleanup(Context context) throws IOException ,InterruptedException {
//保存前5条数据
for( int i = 0; i < tops.length; i++) {
context.write(new LongWritable(tops[i]), NullWritable.get());
}
}
}
//reduce
public static class MyReducer extends Reducer<LongWritable, NullWritable, LongWritable, NullWritable> {
//临时变量
Long temp = Long.MIN_VALUE;
//Top5存储空间
long[] tops;
/** 次方法在run中调用，在全部map之前执行一次 **/
protected void setup(Context context) {
//初始化长度为5
tops = new long[5];
}
//因为每个文件都得到5个值，再次将这些值比对，得到最大的
protected void reduce(LongWritable k2, Iterable<NullWritable> v2, Context context) throws IOException ,InterruptedException {
long top = Long.parseLong(k2.toString());
//
tops[0] = top;
//
Arrays.sort(tops);
}
/** ---此方法在全部到reduce任务结束后执行一次。输出前5个最大值--- **/
protected void cleanup(Context context) throws IOException, InterruptedException {
//保存前5条数据
for( int i = 0; i < tops.length; i++) {
context.write(new LongWritable(tops[i]), NullWritable.get());
}
}
}
// 设置输出类型 job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(NullWritable.class);
本例中的单表实际就是一个文本文件。
//map
public static class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
//拆分原始数据
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//按制表符拆分记录
String[] splits = v1.toString().split("\t");
//一条k2v2记录：把孙辈作为k2；祖辈加下划线区分，作为v2
context.write(new Text(splits[0]), new Text("_"+splits[1]));
//一条k2v2记录：把祖辈作为k2；孙辈作为v2。就是把原两个单词调换位置保存
context.write(new Text(splits[1]), new Text(splits[0]));
}
/**
张三 _张三爸爸
张三爸爸 张三
张三爸爸 _张三爷爷
张三爷爷 张三爸爸
**/
}
//reduce
public static class MyReducer extends Reducer<Text, Text, Text, Text> {
//拆分k2v2[...]数据
protected void reduce(Text k2, Iterable<Text> v2, Context context) throws IOException ,InterruptedException {
String grandchild = ""; //孙辈
String grandfather = ""; //祖辈
/**
张三爸爸 [_张三爷爷，张三]
**/
//从迭代中遍历v2[...]
for (Text man : v2) {
String p = man.toString();
//如果单词是以下划线开始的
if(p.startsWith("_")){
//从索引1开始截取字符串，保存到祖辈变量
grandfather = p.substring(1);
}
//如果单词没有下划线起始
else{
//直接赋值给孙辈变量
grandchild = p;
}
}
//在得到有效数据的情况下
if( grandchild!="" && grandfather!=""){
//写出得到的结果。
context.write(new Text(grandchild), new Text(grandfather));
}
/**
k3=张三，v3=张三爷爷
**/
}
}
// 设置输出类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class);
本例中仍简单采用两个文本文件。
//map
public static class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
//拆分原始数据
protected void map(LongWritable k1, Text v1, Context context) throws IOException ,InterruptedException {
//拆分记录
String[] splited = v1.toString().split("\t");
//如果第一列是数字（使用正则判断），就是地址表
if(splited[0].matches("^[-+]?(([0-9]+)([.]([0-9]+))?|([.]([0-9]+))?)$")){
String addreId = splited[0];
String address = splited[1];
//k2，v2-加两条下划线作为前缀标识为地址
context.write(new Text(addreId), new Text("__"+address));
}
//否则就是人员表
else{
String personId = splited[1];
String persName = splited[0];
//k2，v2-加两条横线作为前缀标识为人员
context.write(new Text(personId), new Text("--"+persName));
}
/**
1 __北京
1 --张三
**/
}
}
//reduce
public static class MyReducer extends Reducer<Text, Text, Text, Text> {
//拆分k2v2[...]数据
protected void reduce(Text k2, Iterable<Text> v2, Context context) throws IOException ,InterruptedException {
String address = ""; //地址
String person = ""; //人员
/**
1, [__北京，--张三]
**/
//迭代的是address或者person
for (Text text : v2) {
String tmp = text.toString();
if(tmp.startsWith("__")){
//如果是__开头的是address
address = tmp.substring(2); //从索引2开始截取字符串
}
if(tmp.startsWith("--")){
//如果是--开头的是person
person = tmp.substring(2);
}
}
context.write(new Text(person), new Text(address));
}
/**
k3=张三，v3=北京
**/
// 设置输出类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class);
关于“Hadoop中MapReduce常用算法有哪些”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，使各位可以学到更多知识，如果觉得文章不错，请把它分享出去让更多的人看到。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
