177 lines
6.3 KiB
Plaintext
177 lines
6.3 KiB
Plaintext
# 直观理解
|
||
|
||
统计一堆牌中有多少张黑桃，最简单的方法是一张一张去数。而 MapReduce 的方法是：
|
||
|
||
1. 把这堆牌分配给多个玩家；
|
||
2. 让每个玩家数自己手中有多少张黑桃；
|
||
3. 把所有玩家数出的黑桃数加起来就是最后的结果。
|
||
|
||
通过让多个玩家来并行地进行统计，MapReduce 可以将统计时间大大缩短。
|
||
|
||
# 大数据
|
||
|
||
从大量数据中快速提取出有价值的信息。注意到这里的数据不一定需要同种类型的数据，可以是多种多样的数据，包括非结构化数据、半结构化数据以及结构化数据。
|
||
|
||
# 基本原理
|
||
|
||
MapReduce 用来对大数据中的海量数据进行离线分析。
|
||
|
||
MapReduce 分为两个阶段：Map 阶段和 Reduce 阶段。
|
||
|
||
在下图中，file 包含多个 block，每个块代表一个海量数据。file 被划分为多个 split，每个分片由一个 Mapper Task 去处理。Map 过程中输入的是 [K1, V1] 数据，这种是一种键值对形式的数据，键为泛型 K1 的类型，值为泛型 V1 的类型。输入也是一个泛型的键值对 [K2, V2]。Shuffle 过程主要是对 Map 阶段输出的键值对进行整合，将键相同的键值对整合到一组中，得到 [K2, {V2,...}] 的整合数据，作为 Reudce 阶段的输入。Reduce 处理完之后得到 [K3, V3] 键值对，Hadoop 会将输出的数据存到分布式文件系统 HDFS 中。
|
||
|
||
|
||
|
||
# WordCount 实战
|
||
|
||
## 开发环境配置
|
||
|
||
创建 Maven 项目，在 pom.xml 中添加以下依赖：
|
||
|
||
```
|
||
<properties>
|
||
<hadoopVersion>3.0.0</hadoopVersion>
|
||
</properties>
|
||
|
||
<dependencies>
|
||
<dependency>
|
||
<groupId>org.apache.hadoop</groupId>
|
||
<artifactId>hadoop-common</artifactId>
|
||
<version>${hadoopVersion}</version>
|
||
</dependency>
|
||
<dependency>
|
||
<groupId>org.apache.hadoop</groupId>
|
||
<artifactId>hadoop-hdfs</artifactId>
|
||
<version>${hadoopVersion}</version>
|
||
</dependency>
|
||
|
||
<dependency>
|
||
<groupId>org.apache.hadoop</groupId>
|
||
<artifactId>hadoop-mapreduce-client-core</artifactId>
|
||
<version>${hadoopVersion}</version>
|
||
</dependency>
|
||
<dependency>
|
||
<groupId>org.apache.hadoop</groupId>
|
||
<artifactId>hadoop-client</artifactId>
|
||
<version>${hadoopVersion}</version>
|
||
</dependency>
|
||
</dependencies>
|
||
```
|
||
|
||
## 文本
|
||
|
||
```html
|
||
the weather is good
|
||
today is good
|
||
good weather is good
|
||
today has good weather
|
||
```
|
||
|
||
将文本的每一行分成一片，每一片的数据提供给 Mapper 进行处理，因此需要 4 个 Mapper。其中 K1 为行号，V1 位该行的字符串。
|
||
|
||
```html
|
||
Split-0: [0, "the weather is good]
|
||
Split-1: [1, "today is good"]
|
||
Split-2: [2, "good weather is good"]
|
||
Split-3: [3, "today has good weather]
|
||
```
|
||
|
||
## Mapper
|
||
|
||
Mapper 类需要提供四个泛型，分别为 K1, V1, K2, V2。
|
||
|
||
```java
|
||
class WordCountMappper extends Mapper<LongWritable, Text, Text, IntWritable> {
|
||
@Override
|
||
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
|
||
String line = value.toString();
|
||
String[] words = line.split(" ");
|
||
for (String word : words) {
|
||
context.write(new Text(word), new IntWritable(1));
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
```html
|
||
Mapper-0: ["the", 1], ["weather", 1], ["is", 1], ["good", 1]
|
||
Mapper-1: ["today", 1], ["is", 1], ["good", 1]
|
||
Mapper-2: ["good", 2], ["weather", 1], ["is", 1]
|
||
Mapper-3: [today", 1], ["has", 1], ["good", 1], ["weater", 1]
|
||
```
|
||
|
||
## Shuffle
|
||
|
||
排序并将拥有相同键的数据整合。
|
||
|
||
```html
|
||
["good", {1, 1, 2, 1}]
|
||
["has", {1}]
|
||
["is", {1, 1, 1}]
|
||
["the", {1}]
|
||
["today", {1, 2}]
|
||
["weater", {1,1}]
|
||
```
|
||
|
||
## Reducer
|
||
|
||
Reducer 同样要指定 4 个泛型：K2, V2, K3, V3
|
||
|
||
```java
|
||
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
|
||
@Override
|
||
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
|
||
Integer count = 0;
|
||
for (IntWritable value : values) {
|
||
count += value.get();
|
||
}
|
||
context.write(key, new IntWritable(count));
|
||
}
|
||
}
|
||
```
|
||
|
||
```html
|
||
Reducer-0: ["good", 5]
|
||
Reducer-1: ["has", 1]
|
||
Reducer-2: ["is", 3]
|
||
Reducer-3: ["the", 1]
|
||
Reducer-4: ["today", 2]
|
||
Reducer-5: ["weather", 3]
|
||
```
|
||
|
||
## Job
|
||
|
||
将项目打包成 jar 包后使用 Hadoop 来运行。
|
||
|
||
```java
|
||
public class WordCountMapReduce {
|
||
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
|
||
Configuration conf = new Configuration();
|
||
// 新建 Job
|
||
Job job = Job.getInstance(conf, "wordcount");
|
||
job.setJarByClass(WordCountMapReduce.class);
|
||
job.setMapperClass(WordCountMappper.class);
|
||
job.setReducerClass(WordCountReducer.class);
|
||
// 设置 Mapper 和 Reducer 的 Key Value 类型
|
||
job.setMapOutputKeyClass(Text.class);
|
||
job.setOutputValueClass(IntWritable.class);
|
||
job.setOutputKeyClass(Text.class);
|
||
job.setOutputValueClass(IntWritable.class);
|
||
// 设置 HDFS 路径
|
||
FileInputFormat.setInputPaths(job, new Path("hdfs://domain:8020/words"));
|
||
FileOutputFormat.setOutputPath(job, new Path("hdfs://domain:8020/words"));
|
||
// 运行
|
||
boolean b = job.waitForCompletion(true);
|
||
if (!b) {
|
||
System.err.println("This task has failed!!!");
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
|
||
# 参考资料
|
||
|
||
- [Hadoop MapReduce 入门](http://www.jikexueyuan.com/course/2686.html)
|
||
|