spark如何实现并发

输入可能以多个文件的形式存储在 HDFS 上，每个 File 都包含了很多块，称为 Block。当 Spark 读取这些文件作为输入时，会根据具体数据格式对应的 InputFormat 进行解析，一般是将若干个 Block 合并成一个输入分片，称为 InputSplit，注意 InputSplit 不能跨越文件。随后将为这些输入分片生成具体的 Task。InputSplit 与 Task是一一对应的关系。随后这些具体的 Task 每个都会被分配到集群上的某个节点的某个 Executor 去执行。

每个节点可以起一个或多个 Executor。

每个 Executor 由若干 core 组成，每个 Executor 的每个 core 一次只能执行一个 Task 。

每个 Task 执行的结果就是生成了目标 RDD 的一个 partiton。

注意: 这里的 core 是虚拟的 core 而不是机器的物理 CPU 核，可以理解为就是 Executor 的一个工作线程。

而 Task 被执行的并发度 = Executor 数目 * 每个 Executor 核数。

至于 partition 的数目：

对于数据读入阶段，例如 sc.textFile，输入文件被划分为多少 InputSplit 就会需要多少初始 Task。

在 Map 阶段 partition 数目保持不变。

在 Reduce 阶段，RDD 的聚合会触发 shuffle *** 作，聚合后的 RDD 的 partition 数目跟具体 *** 作有关，例如 repartition *** 作会聚合成指定分区数，还有一些算子是可配置的。

关于executor和task的概念可以参考 官方文档

本文使用的源码是spark 2.0.0版本

根据类 DAGScheduler 中的 submitMissingTasks 方法可以知道，在stage中会为每个需要计算的partition生成一个task，换句话说也就是每个task处理一个partition。

当task被提交到executor之后，会根据executor可用的cpu核数，决定一个executor中最多同时运行多少个task。在类 TaskSchedulerImpl 的 resourceOfferSingleTaskSet 方法中， CPUS_PER_TASK 的定义为 val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1) ，也就是说默认情况下一个task对应cpu的一个核。如果一个executor可用cpu核数为8，那么一个executor中最多同是并发执行8个task；假如设置 spark.task.cpus 为2，那么同时就只能运行4个task。

在Yarn的NodeManager节点上启动一个map task或者reduce task，在物理上启动的是一个 jvm进程 ；而Spark的task是Executor进程中的一个 线程 。

在已有的 MySQL 服务器之上使用 Apache Spark （无需将数据导出到 Spark 或者 Hadoop 平台上），这样至少可以提升 10 倍的查询性能。使用多个 MySQL 服务器（复制或者 Percona XtraDB Cluster）可以让我们在某些查询上得到额外的性能提升。你也可以使用 Spark 的缓存功能来缓存整个 MySQL 查询结果表。

思路很简单：Spark 可以通过 JDBC 读取 MySQL 上的数据，也可以执行 SQL 查询，因此我们可以直接连接到 MySQL 并执行查询。那么为什么速度会快呢？对一些需要运行很长时间的查询（如报表或者BI），由于 Spark 是一个大规模并行系统，因此查询会非常的快。MySQL 只能为每一个查询分配一个 CPU 核来处理，而 Spark 可以使用所有集群节点的所有核。在下面的例子中，我们会在 Spark 中执行 MySQL 查询，这个查询速度比直接在 MySQL 上执行速度要快 5 到 10 倍。

另外，Spark 可以增加“集群”级别的并行机制，在使用 MySQL 复制或者 Percona XtraDB Cluster 的情况下，Spark 可以把查询变成一组更小的查询（有点像使用了分区表时可以在每个分区都执行一个查询），然后在多个 Percona XtraDB Cluster 节点的多个从服务器上并行的执行这些小查询。最后它会使用map/reduce 方式将每个节点返回的结果聚合在一起形成完整的结果。

---