Spark是一款通用集群计算框架，和Hadoop的MapReduce类似。由于其提供的抽象更简单，性能和功能上比Hadoop强不少。
它已经越来越流行了。
如果是新的项目，或者是为了学习，那么选择Spark完全没问题。
不过对于一些使用了MapReduce的项目来说，迁移就稍微复杂一些了。
Hadoop自身
Hadoop的使用在不断扩大，但同时越来越多的实践证实了MapReduce并不是通用计算范例。
Hadoop的架构本身为其他可能的替代方案提供了场所，比如Impala项目等等。
而对于Hadoop来说，有一部分Hadoop的实现本身和MapReduce本身的抽象并不一致。
- Mapper和Reducer总是使用键值对作为输入输出
- Reducer处理的级别是键
- Mapper和Reducer的对象的生命周期跨越了多个map()和reduce()，同时还支持了setup()和cleanup()
Spark
Spark也是众多替代方案中的一种。
但是对于已经部署在生产环境的项目而言，一句替代方案是不够的，对于一些实时计算的系统更是这样。
好在利用Spark实现类似MapReduce的模型是完全可行的。同时实现本身还可以更简单，并在大部分情况下更快。
对于MapReduce模型本身，使用Spark来实现反而显得更亲近，毕竟Scala的编码风格和API对于本身源于LISP的MapReduce抽象更接近。
键值对和元组
从最基础的例子来看，如果需要计算一个大型文本文件每一行的长度，对于Hadoop MapReduce来说由于输出是键值对，那么就会使用长度作为键，以1为值。
publicclass LineLengthMapper extends Mapper<LongWritable,Text,IntWritable,IntWritable> {
protectedvoidmap(LongWritable lineNumber, Text line, Context context)
throws IOException, InterruptedException {
context.write(new IntWritable(line.getLength()),new IntWritable(1));
}
}
LineLengthMapper由TextInputFormat提供了输入，对于每一个键值对而言，一行为值，而位置为键。
当然，虽然我很少用到，但是我相信有些任务还是需要用到这个键的。
在Spark中，对应的功能看起来就是这样的
lines.map(line => (line.length,1))
Spark的核心抽象是RDD，这里的输入仅仅是字符串，而不是键值对。而Scala中的元组(Tuple)和上面的键值对很类似。
map()的输出是一个(int，int)的元组。当RDD包含元组的时候，可以调用一些额外的方法，比如reduceByKey()，这也是实现MapReduce的一个重要部分。
Spark的reduceByKey
为了得到长度的输出，Hadoop MapReduce还需要一个Reducer来处理。
publicclass LineLengthReducer extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
protectedvoidreduce(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));
}
}
而在Spark中对应的是
val counts =lines.map(line => (line.length,1)).reduceByKey(_ + _)
Spark的API中包含了reduce()方法，不过这个方式是将键值对处理为一个单值，这不是Hadoop MapReduce的行为。相比之下
reduceByKey更接近。
setup() and cleanup()
在MapReduce中，mapper和reducer都可以包含一个setup和cleanup方法。
比如处理数据库的连接
publicclass SetupCleanupMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
private Connection dbConnection;
protectedvoidsetup(Context context) {
dbConnection = ...;
}
protectedvoidcleanup(Context context) {
dbConnection.close();
}
}
而Spark的map()和flatMap()并没有这种东西，我暂时采用的办法是在调用map()前后处理，比如
lines.mapPartitions { valueIterator =>
if (valueIterator.isEmpty) {
Iterator[...]()
}else {
val dbConnection = ...
valueIterator.map { item =>
val transformedItem = ...
if (!valueIterator.hasNext) {
dbConnection.close()
}
transformedItem
}
}
}
其他概念
Hadoop MapReduce中还有一个隐含的概念Combiner，并且还可以一次触发多个输出。
另外一点是MapReduce的Writeable序列化是独有的。
Spark中有一些替代品，比如accumulators，groupBy，而序列化可以考虑Java自身的或者使用Kryo等等。
MapReduce抽象本身是强力的，只不过在Hadoop的实现中有了一些特殊化，同时函数式语言更加胜任这种工作。
参考
RDD
accumulators
