标签：

MapReduce和Spark比较

MapReduce和Spark 介绍

MapReduce

　　第一代 Hadoop MapReduce 是一个在计算机集群上分布式处理海量数据集的软件框架，包括一个 JobTracker 和一定数量的 TaskTracker。运行流程图如图 1 所示：

在最上层有 4 个独立的实体，即客户端、jobtracker、tasktracker 和分布式文件系统。客户端提交 MapReduce 作业；jobtracker 协调作业的运行；jobtracker 是一个 Java 应用程序，它的主类是 JobTracker；tasktracker 运行作业划分后的任务，tasktracker 也是一个 Java 应用程序，它的主类是 TaskTracker。Hadoop 运行 MapReduce 作业的步骤主要包括提交作业、初始化作业、分配任务、执行任务、更新进度和状态、完成作业等 6 个步骤。

Spark 简介

在导入数据时，模型以 block 形式存在于 worker 上，由 driver 向 worker 分发任务，处理完后 work 向 driver 反馈结果。也可在 work 上对数据模型建立高速缓存 cache，对 cache 的处理过程与 block 类似，也是一个分发、反馈的过程。

　　Spark 的 RDD 概念能够取得和专有系统同样的性能，还能提供包括容错处理、滞后节点处理等这些专有系统缺乏的特性。

　　1、迭代算法：这是目前专有系统实现的非常普遍的一种应用场景，比如迭代计算可以用于图处理和机器学习。RDD 能够很好地实现这些模型，包括 Pregel、HaLoop 和 GraphLab 等模型。
　　2、关系型查询：对于 MapReduce 来说非常重要的需求就是运行 SQL 查询，包括长期运行、数小时的批处理作业和交互式的查询。然而对于 MapReduce 而言，对比并行数据库进行交互式查询，有其内在的缺点，比如由于其容错的模型而导致速度很慢。利用 RDD 模型，可以通过实现许多通用的数据库引擎特性，从而获得很好的性能。
　　3、MapReduce 批处理：RDD 提供的接口是 MapReduce 的超集，所以 RDD 能够有效地运行利用 MapReduce 实现的应用程序，另外 RDD 还适合更加抽象的基于 DAG 的应用程序。
　　4、流式处理：目前的流式系统也只提供了有限的容错处理，需要消耗系统非常大的拷贝代码或者非常长的容错时间。特别是在目前的系统中，基本都是基于连续计算的模型，常住的有状态的操作会处理到达的每一条记录。为了恢复失败的节点，它们需要为每一个操作复制两份操作，或者将上游的数据进行代价较大的操作重放，利用 RDD 实现离散数据流，可以克服上述问题。离散数据流将流式计算当作一系列的短小而确定的批处理操作，而不是常驻的有状态的操作，将两个离散流之间的状态保存在 RDD 中。离散流模型能够允许通过 RDD 的继承关系图进行并行性的恢复而不需要进行数据拷贝。

Spark 内部术语解释

MapReduce 转换到 Spark

public class LineLengthCountMapper
 extends Mapper<LongWritable,Text,IntWritable,IntWritable> {
 @Override
 protected void map(LongWritable lineNumber, Text line, Context context)
 throws IOException, InterruptedException {
 context.write(new IntWritable(line.getLength()), new IntWritable(1));
 }
}

lines.map(line => (line.length, 1))

public class LineLengthReducer
 extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
 @Override
 protected void reduce(IntWritable length, Iterable<IntWritable> counts, Context context)
 throws IOException, InterruptedException {
 int sum = 0;
 for (IntWritable count : counts) {
 sum += count.get();
 }
 context.write(length, new IntWritable(sum));
 }
}

val lengthCounts = lines.map(line => (line.length, 1)).reduceByKey(_ + _)

public class CountUppercaseMapper
 extends Mapper<LongWritable,Text,Text,IntWritable> {
 @Override
 protected void map(LongWritable lineNumber, Text line, Context context)
 throws IOException, InterruptedException {
 for (String word : line.toString().split(" ")) {
 if (Character.isUpperCase(word.charAt(0))) {
 context.write(new Text(word), new IntWritable(1));
 }
 }
 }
}

lines.flatMap(
_.split(" ").filter(word => Character.isUpperCase(word(0))).map(word => (word,1))
)

public class CountUppercaseReducer
 extends Reducer<Text,IntWritable,Text,IntWritable> {
 @Override
 protected void reduce(Text word, Iterable<IntWritable> counts, Context context)
 throws IOException, InterruptedException {
 int sum = 0;
 for (IntWritable count : counts) {
 sum += count.get();
 }
 context.write(new Text(word.toString().toUpperCase()), new IntWritable(sum));
 }
}

groupByKey().map { case (word,ones) => (word.toUpperCase, ones.sum) }

reduceByKey(_ + _).map { case (word,total) => (word.toUpperCase,total) }

public class SetupCleanupMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
 private Connection dbConnection;
 @Override
 protected void setup(Context context) {
 dbConnection = ...;
 }
 ...
 @Override
 protected void cleanup(Context context) {
 dbConnection.close();
 }
}

MapReduce程序转换为Spark程序

标签：

原文地址：http://blog.csdn.net/lw_ghy/article/details/51470008