首页 > 其他 > 详细

MapReduce

时间:2019-12-24 21:22:24      阅读:83      评论:0      收藏:0      [点我收藏+]

1.mapreduce定义和优缺点?

MapReduce定义:MapReduce是一个分布式运算的编程框架,使用户开发“基于hadoop的数据分析应用”的核心框架。

优点:

1,MapReduce易于编程

它简单的实现一些接口,就可以完成一个分布式程序,这个分布式程序可以分布到大量廉价的pc机上运行。也就是说写一个分布式程序,跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。

2,良好的扩展性

当你的计算机资源不能得到满足的时候,你可以通过简单的增加机器扩展它的计算能力。

3,高容错性

MapReduce设计的初衷就是使程序能够部署在廉价的pc机器上,这就是要求它具有很高的容错性。比如其中一台机器挂了,它可以把上面的机任务转移到另一个节点上运行,不至于这个任务运行失败,而且这个过程不需要人工参与,而完全是由hadoop内部完成的。

4,适合pb级以上海量数据的离线处理

可以实现上千台服务器集群并发工作,提供数据处理能力。

缺点:

1不擅长实时计算

MapReduce无法像mysql一样,在毫秒或者秒级内返回结果。

2,不擅长流式计算

流式计算的输入数据是动态的,而MapReduce的输入数据集是静态的,不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。

3,不擅长DAG(有向图)计算

多个应用程序存在依赖关系,后一个应用程序的输入为前一个的输出。在这种情况下MapReduce并不是不能做,而是使用后,每个MapReduce作业的输出结果都会写入磁盘,会造成大量的磁盘IO,导致性能非常的低下。

2.mapreduce的数据类型

Java类型

Hadoop Writable类型

Boolean

BooleanWritable

Byte

ByteWritable

Int

IntWritable

Float

FloatWritable

Long

LongWritable

Double

DoubleWritable

String

Text

Map

MapWritable

Array

ArrayWritable

3.查看官方的wordcount代码样例


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

import java.io.IOException;

public class WcDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//获取一个job实例
Job job = Job.getInstance(new Configuration());

//设置本程序的jar包类的路径
job.setJarByClass(WcDriver.class);

//设置map类和reduce类
job.setMapperClass(WcMapper.class);
job.setReducerClass(WcReducer.class);

//设置map输出的key和value类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

//设置Reduce的输出的key和value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

//设置处理文本的输入和输出路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));

//提交job任务
boolean b = job.waitForCompletion(true);
System.exit(b ? 0:1);
}
}

 

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;
//hello word
//hello word
/**
* KEYIN:LongWritable(偏移量)
* VALUEIN:Text (文本中每一行的内容)
* KEYOUT: Text (某一个单词作为key)
* VALUEOUT:IntWritable (单词出现的个数)
*/
public class WcMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
//map输出的key值
private Text keyText = new Text();
//map输出的value值
private IntWritable one = new IntWritable(1);
//ctrl+o
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// super.map(key, value, context);
//获取文件中的行数据
String line = value.toString();
//分割字符串(根据空格分割字符串)
String[] fileds = line.split(" ");
for (String filed : fileds) {
keyText.set(filed);
context.write(keyText,one);
}
}
}

 


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

import java.io.IOException;

/**
* KEYIN: Text map端的输出的key作为reduce的输入key
* VALUEIN: IntWritable map端的输出作为reduce的输入value
* KEYOUT: Text 以单词作为输出的key值
* VALUEOUT:IntWritable 统计单词出现的总数,作为输出的value
*/
public class WcReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable total = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
// super.reduce(key, values, context);
int sum = 0; //统计单词出现的总数 {Text("hadoop"),IntWriter(1),Text("hadoop"),IntWriter(1),Text("hadoop"),IntWriter(1)}
for (IntWritable value : values) {
//累加,统计单词出现总次数
sum += value.get();
}
total.set(sum);
//key,value 写入到上下文中(context)
context.write(key,total);
}
}

4.自己实现wordcount代码

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;
//hello word
//hello word
/**
* KEYIN:LongWritable(偏移量)
* VALUEIN:Text (文本中每一行的内容)
* KEYOUT: Text (某一个单词作为key)
* VALUEOUT:IntWritable (单词出现的个数)
*/
public class WcMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
//map输出的key值
private Text keyText = new Text();
//map输出的value值
private IntWritable one = new IntWritable(1);
//ctrl+o
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// super.map(key, value, context);
//获取文件中的行数据
String line = value.toString();
//分割字符串(根据空格分割字符串)
String[] fileds = line.split(" ");
for (String filed : fileds) {
keyText.set(filed);
context.write(keyText,one);
}
}
}

 

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

import java.io.IOException;

/**
* KEYIN: Text map端的输出的key作为reduce的输入key
* VALUEIN: IntWritable map端的输出作为reduce的输入value
* KEYOUT: Text 以单词作为输出的key值
* VALUEOUT:IntWritable 统计单词出现的总数,作为输出的value
*/
public class WcReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable total = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
// super.reduce(key, values, context);
int sum = 0; //统计单词出现的总数 {Text("hadoop"),IntWriter(1),Text("hadoop"),IntWriter(1),Text("hadoop"),IntWriter(1)}
for (IntWritable value : values) {
//累加,统计单词出现总次数
sum += value.get();
}
total.set(sum);
//key,value 写入到上下文中(context)
context.write(key,total);
}
}

 

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

import java.io.IOException;

public class WcDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//获取一个job实例
Job job = Job.getInstance(new Configuration());

//设置本程序的jar包类的路径
job.setJarByClass(WcDriver.class);

//设置map类和reduce类
job.setMapperClass(WcMapper.class);
job.setReducerClass(WcReducer.class);

//设置map输出的key和value类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

//设置Reduce的输出的key和value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

//设置处理文本的输入和输出路径
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));

//提交job任务
boolean b = job.waitForCompletion(true);
System.exit(b ? 0:1);
}
}

MapReduce

原文:https://www.cnblogs.com/lu0420-0412/p/12093350.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!