本文翻译自
Apache Spark: core concepts, architecture and internals

本文覆盖了Apache Spark的RDD、DAG、执行工作流、tasks的stages的形成、shuffle的实现等核心概念，还描述了Spark的架构和它的主要模块Spark Driver。

介绍

Spark是分布式数据处理的通用框架，提供了用于大规模操作数据的功能API，内存数据缓存和可重用计算。它将一组coarse-grained转换作用于分区数据，如果失败则利用数据集的血统来重新计算tasks。值得一提的是，Spark支持大多数数据格式，与各种存储系统集成，可以在Mesos或YARN上执行。

功能强大且简洁的API与丰富的库相结合，可以更轻松地大规模执行数据操作。例如，以Parquet格式执行Cassandra列族的备份和恢复：

def backup(path: String, config: Config) {  
  sc.cassandraTable(config.keyspace, config.table)
    .map(_.toEvent).toDF()
    .write.parquet(path)
}

def restore(path: String, config: Config) {  
  sqlContext.read.parquet(path)
  .map(_.toEvent)
  .saveToCassandra(config.keyspace, config.table)
}

或运行差异分析，比较不同数据存储中的数据：

sqlContext.sql {  
  """
     SELECT count()
     FROM cassandra_event_rollups
     JOIN mongo_event_rollups
     ON cassandra_event_rollups.uuid = cassandra_event_rollups.uuid
     WHERE cassandra_event_rollups.value != cassandra_event_rollups.value
  """.stripMargin
}

概述

Spark围绕弹性分布式数据集（RDD）和有向无环图（DAG）的概念构建，DAG表示它们之间的转换和依赖。

Spark-Overview

Spark Application（通常也称为Driver Program或者Application Master）在高级别由SparkContext和用户代码组成，用户代码与它交互创建RDD并执行一系列转换以实现最终结果。然后将这些RDD的转换变为DAG并提交给Scheduler以在一组工作节点上执行。

RDD：Resilient Distributed Dataset
RDD可以被认为是具有故障恢复可能性的不可变并行数据结构。它提供了各种对数据进行转换和实体化（materializations ），以及控制元素的缓存和分区以优化数据放置的API。RDD既可以从外部存储创建，也可以从另一个RDD创建，并存储有关其父节点的信息，以优化执行（通过流水线操作），并在发生故障时重新计算分区。

从开发者的角度来看，RDD表示分布式的不可变数据（分区数据+iterator），且存在惰性机制（transformations）。RDD接口定义了五个主要的属性：

//a list of partitions (e.g. splits in Hadoop)
def getPartitions: Array[Partition]

//a list of dependencies on other RDDs
def getDependencies: Seq[Dependency[_]]

//a function for computing each split
def compute(split: Partition, context: TaskContext): Iterator[T]

//(optional) a list of preferred locations to compute each split on
def getPreferredLocations(split: Partition): Seq[String] = Nil

//(optional) a partitioner for key-value RDDs
val partitioner: Option[Partitioner] = None  

这是一个调用sparkContext.textFile("hdfs://...")方法创建RDD的例子：
首先加载HDFS blocks到内存，然后使用map()函数过滤keys，创建两个RDDS：

DAG-logical-vs-partitions-view

• HadoopRDD
  • getPartitions = HDFS blocks
  • getDependencies = None
  • compute = load block in memory
  • getPrefferedLocations = HDFS block locations
  • partitioner = None
• MapPartitionsRDD
  • getPartitions = same as parent
  • getDependencies = parent RDD
  • compute = compute parent and apply map()
  • getPrefferedLocations = same as parent
  • partitioner = None

RDD操作

RDD的操作分为以下几种：

• Transformations
  • 将用户函数应用于分区中的每个元素（或整个分区）
  • 将聚合函数应用于整个数据集（groupBy，sortBy）
  • 引入RDD之间的依赖关系以形成DAG
  • 提供重新分区的功能（repartition，partitionBy）
• Actions
  • 触发job执行
  • 用于实现计算结果
• Extra: persistence
  • 显式地将RDD存储在内存，磁盘或堆外(off-heap)（cache, persist）
  • 检查点（checkpoint），截断RDD的血统

下面是一些代码示例，其中汇总了来自Cassandra的lambda样式的数据，将先前汇总的数据与原始存储中的数据相结合，并演示了RDD上可用的一些转换和操作

//aggregate events after specific date for given campaign
val events =  
    sc.cassandraTable("demo", "event")
      .map(_.toEvent)                                
      .filter { e =>
        e.campaignId == campaignId && e.time.isAfter(watermark)
      }
      .keyBy(_.eventType)
      .reduceByKey(_ + _)                                        
      .cache()                                            

//aggregate campaigns by type
val campaigns =  
    sc.cassandraTable("demo", "campaign")
      .map(_.toCampaign)
      .filter { c => 
         c.id == campaignId && c.time.isBefore(watermark)
      }
      .keyBy(_.eventType)
      .reduceByKey(_ + _)
      .cache()

//joined rollups and raw events
val joinedTotals = campaigns.join(events)  
           .map { case (key, (campaign, event)) => 
             CampaignTotals(campaign, event) 
            }
           .collect()

//count totals separately
val eventTotals =  
    events.map{ case (t, e) => s"$t -> ${e.value}" }
    .collect()

val campaignTotals =  
    campaigns.map{ case (t, e) => s"$t -> ${e.value}" }
    .collect()

执行工作流概述

Spark-Scheduling-Process

执行工作流：将包含RDD转换的用户代码变成有向无环图，然后由DAGScheduler划分stages。stagese组合了不需要shuffling/repartitioning数据的任务。tasks运行在workers上，然后将结果返回客户端。

DAG

Logical-view

这是上面示例代码的DAG。因此，基本上任何数据处理工作流都可以定义为读取数据源，应用一组转换并以不同方式实现结果。
转换在RDD之间创建依赖关系，在这里我们可以看到它们的不同类型。

依赖关系通常分为“窄”和“宽”：

Dependency-Types

• 具有“窄”依赖的RDD操作，如map()和filter()

Spark中有两种类型的tasks：ShuffleMapTask将其输入分区，ResultTask将其输出发送给driver。
两种类型的stages：ShuffleMapStage和ResultStage。

Shuffle

在shuffle期间，ShuffleMapTask将blocks写入本地文件，然后接下来的stages中的tasks通过网络抓取这些blocks。

• Shuffle Write

  • 在分区之间重新分配数据并将文件写入磁盘
  • 每个hash shuffle task为每个“reduce” task创建一个文件（total = MxR）
  • sort shuffle task创建一个文件，其中区域分配给reducer
  • sort shuffle使用内存排序和溢出到磁盘以获得最终结果

• Shuffle Read

  • 获取文件并应用reduce()逻辑
  • 如果需要数据有序，则对于任何类型的shuffle，它在“reducer”侧排序

在Spark Sort Shuffle是自1.2以来的默认值，但Hash Shuffle也可用。

Sort Shuffle

Sort-Shuffle

• 传入记录根据其目标分区ID在内存中累加和排序
• 如果溢出，已排序的记录将写入一个或多个文件，然后合并
• 索引文件存储数据文件中数据块的偏移量
• 在某些条件下，可以在不进行反序列化的情况下进行排序 SPARK-7081

Spark组件

Spark有三个主要的组件：

Spark-Cluster-Architecture

• Spark Driver
  • 用来执行用户应用程序的单独进程
  • 创建SparkContext以计划jobs执行并与集群管理器协商
• Executors
  • 运行driver安排的tasks
  • 将计算结果存储在内存，磁盘或堆外
  • 与存储系统交互
• Cluster Manager
  • Mesos
  • YARN
  • Spark Standalone

相关阅读：

[1] Apache Spark: core concepts, architecture and internals
[2] SparkInternals 对Spark的深刻解读，非常好的内容！
[3] Spark shuffle introduction