1. Spark Core 进阶

注意：

(1)在sparksql、sparkstreaming、sparkml或者sparkmllib的底层，都是触发Spark Core 计算的，Spark Core它从整个计算过程的调度需要通过我们编写的driver程序来进行驱动，因为driver驱动程序会放到集群中去，从而触发Spark Core计算

(2)Executor可以类似理解为一个虚拟机或者容器，是具体执行task的单元，从而进行干活，默认会分配1G的内存

注意：

(1)做资源调度的组件有YARN集群、Standalone(图上的两种)。对于YARN集群来说的话，RM是负责全局资源的，而NM是进行向RM申 请资源的(要求分配内存资源)，如果RM中有多余资源的话，就可以进行分配资源给NM(RM的资源在NM中才能够进行运行)，然后NM会有一个executer来进行干活，进而当NM进行干活时会有一个AM，AM起到了一个对NM监控的作用(监管它任务有没有做完),当NM任务结束后，AM也会也会消失,此时进行一个对资源的回收。RM和AM之间有一个心跳包，当NM停止运行后，RM和AM心跳包自然消失，RM后按以上的形式递归下去进行，会进行给下一个NM分配资源。整个过程是通过YARN来进行管理的。

(2)DAG就是用来构建stage(DAG图构建出来就是来进行划分stage的，stage就是根据DAG图来进行划分stage的，主要原理是DAG将rdd进行拆解来划分stage)，使得各stage之间是没有依赖的，从而可以使得各stage之间可以做并行计算。在DAG内部都是一些窄依赖，在stage之间是一种宽依赖。

(3)一个stage可以由多台机器进行运行，达到stage中任务执行的并行运行

(4)自己要弄清楚Worker Node、Executor、RDD、DAG、stage之间的关系

1.1 Spark workflow

1.1.1 RDD

RDD 是 Spark 最核心的部分，spark 将所有数据都抽象成 RDD

1.1.2 Scheduler

Spark 的调度机制，分为 DAGScheduler 和 TaskScheduler

1.1.3 Storage

最要管理缓存以后的 RDD、shuffle 中间结果数据和 broadcast 数据

1.1.4 Shuffle

Shuffle 分为 Hash 方式和 Sort 方式，两种方式的中间数据都写本地盘

1.2 部署及优化

注意： 在spark里面进行部署集群有3种方式：a、local() b、standalone-集群----将文件存储在本地 c、yarn/mesds集群*

1.2.1 本地模式

用于开发和单元测试，在本地通过在 JVM 中运行多线程模拟分布式并行计算，例如：--master local[4]

1.2.2 Spark on YARN的基本架构和工作流程

1.2.3 Standalone集群

1.2.4 优化

开发调优：

1、避免创建重复的 RDD；

2、尽可能复用 RDD；

3、对多次使用的 RDD 进行持久化；

4、尽量避免使用 Shuffle；

5、优化算子，利用reducebyKey 代替 groupbyKey、mapPartitions 代替 map、foreachPartitions代替 foreach、filter 之后 coalesce 操作实现 partitions 的压缩、使用repartitionAndSortWithinPartitions 替代 repartition 和 sort 操作；

6、Spark同时支持使用 Kryo 序列化库，Kryo 序列化类库的性能比 Java 序列化类库的性能要高很多；

7、尽量使用字符串替代对象，使用原始类型(比如Int、Long)替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低 GC 频率，提升性能；

资源调优：Executor 的内存主要分为三块：第一块是让 task 执行我们自己编写的代码时使用，默认是占 Executor 总内存的 20%；第二块是让 task 通过 shuffle 过程拉取了上一个 stage 的 task 的输出后，进行聚合等操作时使用，默认也是占 Executor 总内存的 20%；第三块是 让 RDD 持久化时使用，默认占 Executor 总内存的 60%。task 的执行速度是跟每个 Executor 进程的CPU core 数量有直接关系的。一个 CPU core 同一时间只能执行一个线程。而每个Executor 进程上分配到的多个 task，都是以每个 task 一条线程的方式，多线程并发运行的。如果 CPU core 数量比较充足，而且分配到的 task数量比较合理，那么通常来说，可以比较快速和高效地执行完这些 task 线程。

资源参数调优(shuffle 参数调优)：资源参数的调优，没有一个固定的值，需要在业务环境中测试获得。

数据倾斜调优(先进行数据倾斜，后做优化)：两阶段聚合(局部、全局)

1.3 Spark 常用 API 介绍

参考文档：http://spark.apache.org/docs/latest/api/python/index.html

参考文档：http://spark.apache.org/docs/latest/ml-guide.html

常见的转换操作(Transformation API)：

• filter(func)：筛选出满足函数 func 的元素，并返回一个新的数据集

• map(func)：将每个元素传递到函数 func 中，并将结果返回为一个新的数据

集

• flatMap(func)：与 map()相似，但每个输入元素都可以映射到 0 或多个输出

结果

• groupByKey()：应用于(K,V)键值对的数据集时，返回一个新的(K, Iterable)形

式的数据集

• reduceByKey(func)：应用于(K,V)键值对的数据集时，返回一个新的(K, V)形式的数据集，其中的每个值是将每个 key 传递到函数 func 中进行聚合

行动操作是真正触发计算的地方。Spark 程序执行到行动操作时，才会执行真正的计算，从文件中加载数据，完成一次又一次转换操作，最终，完成行动操作得到结果。

下面列出一些常见的行动操作(Action API)：

• count() 返回数据集中的元素个数

• collect() 以数组的形式返回数据集中的所有元素

• first() 返回数据集中的第一个元素

• take(n) 以数组的形式返回数据集中的前 n 个元素

• reduce(func) 通过函数 func(输入两个参数并返回一个值)聚合数据集中的元素

• foreach(func) 将数据集中的每个元素传递到函数 func 中运行

1.4 实例分析

蒙特卡罗计算圆周率

2. Spark 上的查询

2.1 SparkSQL

在 Shark 原有的架构上重写了逻辑执行计划的优化部分，解决了 Shark 存在的问题。Spark SQL 在 Hive 兼容层面仅依赖 HiveQL 解析和 Hive 元数据，也就是说，从 HQL被解析成抽象语法树(AST)起，就全部由 Spark SQL 接管了。Spark SQL 执行计划生成和优化都由 Catalyst(函数式关系查询优化框架)负责。

DataFrame

RDD 是分布式的 Java 对象的集合，比如，RDD[Person]是以 Person 为类型参数，但是，Person 类的内部结构对于 RDD 而言却是不可知的。DataFrame 是一种以 RDD 为基础的分布式数据集，也就是分布式的 Row 对象的集合(每个 Row 对象代表一行记录)，提供了详细的结构信息，也就是我们经常说的模式(schema)，Spark SQL 可以清楚地知道该数据集中包含哪些列、每列的名称和类型。

和 RDD 一样，DataFrame 的各种变换操作也采用惰性机制，只是记录了各种转换的逻辑转换路线图(是一个 DAG 图)，不会发生真正的计算，这个 DAG 图相当于一个逻辑查询计划，最终，会被翻译成物理查询计划，生成 RDD DAG，按照之前介绍的RDD DAG 的执行方式去完成最终的计算得到结果。

注意：

(1)pandas中的dataframe和sparksql中的dataframe是不同的，sparksql中的dataframe是基于rdd的一个封装，是sparksql基本数据类型，所以具有分布式结构的特性，两者都支持索引。

(2)pandas中的dataframe通过异构数据转换转为sparksql中的dataframe，会涉及到Apache Arrow

(3)rdd只区分弹性式分布数据集，不存在表结构，没有数据类型；dataframe存在表结构，并且可以通过索引结构可以来查找相应的行和列；在两者之间存在一个dataset,它也存在表结构，是需要数据类型，dataframe是dataset的一个具体实例

操作

1、df.printSchema()//打印出表结构，相当于mysql中的describle 表名一样

2、df.select(df("name"),df("age")+1).show()

3、df.filter(df("age")>20).show()

4、df.groupBy("age").count().show()

5、df.sort(df("age").desc).show()

6、df.sort(df("age").desc,df("name").asc).show()

7、df.select(df("name").as("username"),df("age")).show()

Parquet 格式数据(是一个序列化之后的数据)

Parquet 是一种流行的列式存储格式，可以高效地存储具有嵌套字段的记录。Parquet 是语言无关的，而且不与任何一种数据处理框架绑定在一起，适配多种语言和组件，能够与 Parquet 配合的组件有：

• 查询引擎:Hive,Impala, Pig, Presto,Drill,Tajo,HAWQ, IBM Big SQL

• 计算框架: MapReduce, Spark, Cascading, Crunch, Scalding, Kite

• 数据模型:Avro,Thrift,Protocol Buffers,POJO

知识点补充：

(1)df.write.parquet(“XXXX”)是将rdd类型的dataframe转化为Parquet 格式数据存储到hdfs中

2.2. Hive on spark

Shark 即 Hive on Spark，为了实现与 Hive 兼容，Shark 在 HiveQL 方面重用了 Hive中的 HiveQL 解析、逻辑执行计划翻译、执行计划优化等逻辑，可以近似认为仅将物理执行计划从 MapReduce 作业替换成了 Spark 作业，通过 Hive 的 HiveQL 解析，把HiveQL 翻译成 Spark 上的 RDD 操作。

Shark 的设计导致了两个问题：一是执行计划优化完全依赖于 Hive，不方便添加新的优化策略；二是因为 Spark 是线程级并行，而MapReduce 是进程级并行，因此，Spark 在兼容 Hive 的实现上存在线程安全问题，导致 Shark 不得不使用另外一套独立维护的打了补丁的 Hive 源码分支。

Shark 的实现继承了大量的 Hive 代码，因而给优化和维护带来了大量的麻烦，特别是基于 MapReduce 设计的部分，成为整个项目的瓶颈。因此，在 2014 年的时候，Shark 项目中止，并转向 Spark SQL 的开发。

注意：

1)原生的hive是在hadoop上的mapreduce来进行相应的运算的，在spark上的hive是替换掉了原来在hadoop上的mapreduce，即用的是spark上的内存mapreduce( Hive on Spark 是原生Hive进行兼容操作(原生的Hive是Hadoop上的外存mapreduce)，其本质就是将Hadoop上的mapreduce转化为spark上的mapreduce。spark SQL 比Hive on Spark更加有优势，Hive on Spark只有兼容操作优势。(spark sql 的优化更加彻底、更加先进))

2)Hive on spark不等同于SparkSQL，前者只是做了兼容的操作

3)hive是非内存式的查询,SparkSQL是内存式的查询

4)hive就是把sql语句转化成mapreduce(MR)对象，从而进行计算

5)Hive on spark 就是把hadoop底层的集群替换成spark底层的集群，用spark的方式实现简单的兼容

补充知识点

(1) 数据库和数据仓库：a、数据库是是oltp模式(在线事务)，用来进行在线数据查询和进行增删改查。 b、数据仓库是进行聚合和分析的(即mpp+聚合分析)，数据仓库遵从的是olap模式(在线分析)(Atheana和bigquery属于的是数据仓库，其针对的是非结构化数据库)

(2)Neo4j是图存储数据库，专门存储图的。

(3)bigtable的开源实现是hbase

(4)blob数据类型(在字符串的数据类型中)是存储二进制数据(主要是.mp3、.mp4、.jpeg等文档格式的二进制数据的的文档的格式文件)

(5)hive(只是一个查询工具，不具有存储结构)和hbase是hadoop之上的应用，而hadoop的底层应用是mapreduce

(6)SparkSQL是内存式数据库