关键词：
关于MR的工作原理不做过多叙述，本文将对MapReduce的实例WordCount(单词计数程序)做实践，从而理解MapReduce的工作机制。
WordCount:
1.应用场景，在大量文件中存储了单词，单词之间用空格分隔
2.类似场景：搜索引擎中，统计最流行的N个搜索词，统计搜索词频率，帮助优化搜索词提示。
3.采用MapReduce执行过程如图
3.1MapReduce将作业的整个运行过程分为两个阶段
3.1.1Map阶段和Reduce阶段
Map阶段由一定数量的Map Task组成
输入数据格式解析：InputFormat
输入数据处理：Mapper
数据分组：Partitioner
3.1.2Reduce阶段由一定数量的Reduce Task组成
数据远程拷贝
数据按照key排序
数据处理：Reducer
数据输出格式：OutputFormat
4.介绍代码结构
4.1 pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>hadoop</groupId>
<artifactId>hadoop.mapreduce</artifactId>
<version>1.0-SNAPSHOT</version>
<repositories>
<repository>
<id>aliyun</id>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-client</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>2.7.3</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.3</version>
<configuration>
<classifier>dist</classifier>
<appendAssemblyId>true</appendAssemblyId>
<descriptorRefs>
<descriptor>jar-with-dependencies</descriptor>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
4.2 WordCount.java
package hadoop.mapreduce;
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;
import java.io.IOException;
public class WordCount {
public static class WordCountMap
extends Mapper<Object, Text, Text, IntWritable> {
public void map(Object key,Text value, Context context) throws IOException, InterruptedException {
//在此处写map代码
String[] lines = value.toString().split(" ");
for (String word : lines) {
context.write(new Text(word), new IntWritable(1));
}
}
}
public static class WordCountReducer
extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//在此处写reduce代码
int count=0;
for (IntWritable cn : values) {
count=count+cn.get();
}
context.write(key, new IntWritable(count));
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
System.err.println("Usage: wordcount <in> [<in>...] <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
//设置输入路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
//设置输出路径
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//设置实现map函数的类
job.setMapperClass(WordCountMap.class);
//设置实现reduce函数的类
job.setReducerClass(WordCountReducer.class);
//设置map阶段产生的key和value的类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//设置reduce阶段产生的key和value的类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//提交job
job.waitForCompletion(true);
for (int i = 0; i < otherArgs.length - 1; ++i) {
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
4.3 data目录下文件内容：
to.txt
hadoop spark hive hbase hive
t1.txt
hive spark mapReduce spark
t2.txt
sqoop spark hadoop
5. 数据准备
5.1 maven 打jar包为hadoop.mapreduce-1.0-SNAPSHOT.jar，传入master服务器上
5.2 将需要计算的数据文件放入datajar/in （临时目录无所谓在哪里）
5.3 启动hadoop ，关于hadoop安装可参考我写的文章 大数据系列之Hadoop分布式集群部署
将datajar/in文件传至hdfs 上
hadoop fs -put in /in
#查看文件
hadoop fs -ls -R /in
5.4 执行jar
两种命令方式
#第一种：hadoop jar hadoop jar hadoop.mapreduce-1.0-SNAPSHOT.jar hadoop.mapreduce.WordCount /in/* /out #OR #第二种：yarn jar yarn jar hadoop.mapreduce-1.0-SNAPSHOT.jar hadoop.mapreduce.WordCount /in/* /yarnOut
5.5.执行后输出内容分别如图
hadoop jar ...结果
yarn jar ... 结果
6.查看结果内容
#查看hadoop ja 执行后输出结果目录 hadoop fs -ls -R /out #查看yarn jar 执行后输出结果目录 hadoop fs -ls -R /yarnOut
目录说明：目录中_SUCCESS 是日志文件，part-r-00000是计算结果文件
查看计算结果
#查看out/part-r-00000文件 hadoop fs -text /out/part-r-00000 #查看yarnOut/part-r-00000文件 hadoop fs -text /yarnOut/part-r-00000
完~~~,Java代码内容已上传至GitHub https://github.com/fzmeng/MapReduceDemo
大数据系列之分布式大数据查询引擎presto
... http://prestodb-china.comPRESTO是什么？Presto是一个开源的分布式SQL查询引擎，适用于交互式分析查询，数据量支持GB到PB字节。Presto的设计和编写完全是为了解决像Facebook这样规模的商业数据仓库的交互式分析和处理速度的问题。... 查看详情
google云计算原理与应用之分布式数据处理mapreduce
...模式思想最早是在1995年提出的。MapReduce的特点:与传统的分布式程序设计相比，MapReduce封装了并行处理、容错处理、本地化计算、负载均衡等细节，还提供了一个简单而强大的接口。MapReduce把对数据集的大规模操作，分发给一个... 查看详情
奇点云数据中台技术汇|datasimba系列之计算引擎篇
随着移动互联网、云计算、物联网和大数据技术的广泛应用，现代社会已经迈入全新的大数据时代。数据的爆炸式增长以及价值的扩大化，将对企业未来的发展产生深远的影响，数据将成为企业的核心资产。如何处理大数据，挖... 查看详情
系统设计分布式处理之mapreduce
方法介绍MapReduce是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。这样做的好处是可以在任务被分解后，可以通过大量机器进行并行计算，减少整个操作的时间。... 查看详情
大数据与hadoop之间是啥关系
...，Hadoop是目前常见的大数据支撑性平台，Hadoop平台提供了分布式存储（HDFS）、分布式计算（MapReduce）、任务调度（YARN）、对象存储（Ozone）和组件支撑服务（Common）。参考技术AHadoop，Spark和Storm是目前最重要的三大分布式计算系... 查看详情
大数据入门-分布式计算框架mapreduce
1概述源自于Google的MapReduce论文，发表于2004年12月。HadoopMapReduce是GoogleMapReduce的克隆版优点海量数量离线处理易开发易运行缺点实时流式计算2MapReduce编程模型wordcount词频统计MapReduce执行流程将作业拆分成Map阶段和Reduce阶段Map... 查看详情
大数据处理为何选择spark？
...的计算引擎。2、HadoopHadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下，开发分布式程序。充分利用集群的威力进行高速运算和存储。Hadoop的框架最核心的设计就是：HDFS和MapRedu... 查看详情
大数据之spark
...法Spark是Scala编写，方便快速编程Spark与MR的区别都是分布式计算框架，Spark计算中间结果基于内存缓存，MapReduce基于HDFS存储也正因此，Spark处理数据的能力一般是MR的三到五倍以上Spark中除了基于内存计算这一个计... 查看详情
五种大数据处理架构
...点中提取信息和见解的过程。下文将介绍这些框架：·仅批处理框架：ApacheHadoop·仅流处理框架：ApacheStormApacheSamza·混合框架：ApacheSparkApacheFlink大数据处理框架是什么？处理框架和处理引擎负责对数据系统中的数据进行计算。虽... 查看详情
大数据分布式计算系统spark入门核心之rdd
作者：幻好概述ApacheSpark是一个快速且通用的集群计算系统。提供Java、Scala、Python和R中的高级API，以及支持通用执行图的优化引擎。它还支持一组丰富的高级工具，包括用于SQL和结构化数据处理的SparkSQL、用于机器学习的MLlib、... 查看详情
什么是flink(流处理框架)(代码片段)
Flink入门什么是FlinkApacheFlink是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架Flink官网Flink的发展历史d Flink诞生于欧洲的一个大数据研究项目StratoSphere。该项目是柏林工业大学的一个研究性项... 查看详情
数据湖：分布式开源处理引擎spark
系列专题：数据湖系列文章1.什么是Spark ApacheSpark是一种高效且多用途的集群计算平台。换句话说，Spark是一种开源的，大规模数据处理引擎。它提供了一整套开发API，包括流计算、机器学... 查看详情
数据湖：分布式开源处理引擎spark
系列专题：数据湖系列文章1.什么是Spark ApacheSpark是一种高效且多用途的集群计算平台。换句话说，Spark是一种开源的，大规模数据处理引擎。它提供了一整套开发API，包括流计算、机器学... 查看详情
大数据技术之hadoop（mapreduce）
...pReduce框架原理InputFormat数据输入MapReduce定义MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分... 查看详情
flink编程入门
...内存计算的飞速发展。Spark的火热或多或少的掩盖了其他分布式计算的系统身影。就像Flink，也就在这个时候默默的发展着。在国外一些社区，有很多人将大数据的计算引擎分成了4代，当然，也有很多人不会认同。我们先姑且这... 查看详情
hadoop和mapreduce是一种啥关系？
参考技术Ahadoop是依据mapreduce的原理，用Java语言实现的分布式处理机制。Hadoop是一个能够对大量数据进行分布式处理的软件框架，实现了Google的MapReduce编程模型和框架，能够把应用程序分割成许多的小的工作单元，并把这些单元... 查看详情
《从0开始学大数据》之mapreduce计算框架是如何运作的(代码片段)
MapReduce编程模型将大数据计算过程切分为Map和Reduce两个阶段，在Map阶段为每个数据块分配一个Map计算任务，然后将所有map输出的Key进行合并，相同的Key及其对应的Value发送给同一个Reduce任务去处理。通过这两个阶段... 查看详情
一文带你了解大数据技术之mapreduce
...educe编程规范8.WordCount案例实操1.MapReduce定义MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的 查看详情