Spark中的算子都有哪些

总的来说,spark分为三大类算子:

• Transformation 转换算子：一个RDD 转换生成一个新的 RDD 的操作。 一个RDD 转换生成另一个 RDD 的转换操作不是马上执行，不触发提交作业，需要等到有 Action 操作的时候才会真正触发运算。通过谱系图（lineage）这个DAG图进行操作，进行恢复。

• Action 行动算子：这类算子会触发 SparkContext 提交 Job 作业； Action 算子会触发 Spark 提交作业（Job），对RDD 进行实际计算，并将求得的结果返回到驱动程序中或者写入到外部存储系统中。

• Controller 控制操作：Spark中控制算子也是懒执行的，当我们每次调用行动操作时，都会重算RDD的所有依赖，如果多次行动操作使用同一个RDD，就会导致大量的重复运算。为避免这种现象，可以对数据进行持久化，也就是存储该RDD，保存在各自的分区中。

控制算子有三种，cache,persist（RDD 持久化原理）checkpoint，以上算子都可以将RDD持久化，持久化的单位是partition。
cache和persist都用于将一个RDD进行缓存，这样在以后使用的过程中就不需要重复计算 。cache 默认缓存级别是MEMORY_ONLY
checkpoint算子不仅能将RDD持久化到磁盘，还能切断RDD之间的依赖关系。

RDD工作流📌

👉输入：定义RDD，在 Spark 程序运行中，数据从外部数据空间（例如， HDFS、 Scala 集合或数据）输入到 Spark，数据就进入了 Spark 运行时数据空间，会转化为 Spark 数据块，形成RDD。

👉运行：在 Spark 数据输入形成 RDD 后，进行相应的操作，通过行动（Action）算子，触发 Spark 提交作业，提交作业。如果数据需要复用，可以通过 Cache 算子，将数据缓存到内存。

👉输出：程序运行结束数据会输出 Spark 运行时空间，存储到分布式存储中（如saveAsTextFile 输出到 HDFS）或Scala数据或集合中（ collect 输出到 Scala 集合，count 返回 Scala Int 型数据）。

Spark运行模式(资源调度框架的使用，了解)📌

1. Local模式: 是用单机的多线程或者多进程模拟Spark分布式计算，通常用来 单机调试,验证开发出来的应用程序逻辑上有没有问题。

该方式下，在程序执行过程中，只会生成一个SparkSubmit进程。这个SparkSubmit进程又当爹、又当妈，既是客户提交任务的Client进程、又是Spark的driver程序、还充当着Spark执行Task的Executor角色。其中N代表可以使用N个线程，每个线程拥有一个core。如果不指定N，则默认是1个线程（该线程有1个core）。如果是local[*]，则代表 Run Spark locally with as many worker threads as logical cores on your machine.

2. 分布式部署模式:

👉Standalone模式: 独立模式，自带完整的模式 , 该方式适用master和worker进程来模拟集群形式，不需要启动hadoop。在架构上和 MapReduce1比较，具有一致性，都是由Master、worker组成(只是名称不一样)，资源抽象为粗粒式的slot，多少slot多少task。

👉Spark on YARN：因为现在企业用到 hadoop是基于YARN 的，为了融合Spark，该方式是使用YARN来做集群管理，为Spark应用分配资源 , 进行统一资源管理。有两种方式， YARN-client(用于交互,client当中运行sparkContext进程进行任务分发监控)，YARN-cluster 任务的分发和监控放在 MRAPPmaster当中。

👉 Spark on mesos：YARN和 mesos都是统一资源管理和调度系统。mesos支持粗粒式和细粒式调度，前者节省了资源调度时间的开销，后者是不存在资源的浪费，但是资源调度延迟较大。

讲一下Spark 的运行架构

👉Driver：这是监督 Spark 作业或程序端到端执行的主程序。 它与集群的资源管理器进行资源的协商，并将程序编排成尽可能小的数据本地并行编程单元。

👉Executors：在任何 Spark 任务中，可以有一个或多个 executor，即执行由 drive 委派的较小任务的进程。 executor 处理数据，最好是本地节点的，并将结果存储在内存和 / 或磁盘中。

👉Master：Apache Spark 已经在主 / 从架构中实现，因此 master 指的是执行驱动程序的集群节点。

👉Slave(已经改名为 Worker)：在分布式集群模式下，slave 是指运行执行程序的节点，因此在群集中可以有多个从机（而且大部分情况都是这样）。

👉Job：这是对任何一组数据执行的操作的集合。 典型的 word count job 涉及从任意来源读取文本文件，然后分离 (splitting) 并聚合 (aggregating) 这些字。

👉DAG：Spark 引擎中的任何 Spark 工作都由 DAG 的操作代表。 DAG 按顺序表示 Spark 操作的逻辑执行。 在发生故障的情况下由 DAG 重新计算可能的血统（lineage）。

👉Task：一个 job 可以拆分成更小的单位，以被称为 task 的孤立任务进行操作。 每个 task 由 executor 在一个数据分区上执行。

👉Stages：Spark 作业可以按逻辑划分为多个 stage，每个stage代表一组具有相同的洗牌（shuffle）依赖关系的任务，即发生数据洗牌 (shuffle) 的任务。 在洗牌映射（shuffle map）阶段，任务结果被输入到下一个阶段；在结果阶段，task 计算 action，开始对 Spark job 的赋值，例如take（），foreach（）和 collect（）。

（* 注：shuffle 是划分 DAG 中 stage 的标识, 同时影响 Spark 执行速度的关键步骤。RDD 的 Transformation 函数中, 又分为窄依赖 (narrow dependency) 和宽依赖 (wide dependency) 的操作. 窄依赖跟宽依赖的区别是是否发生 shuffle(洗牌) 操作. 宽依赖会发生shuffle操作. 窄依赖是子RDD的各个分片 (partition) 不依赖于其他分片, 能够独立计算得到结果, 宽依赖指子 RDD 的各个分片会依赖于父 RDD 的多个分片, 所以会造成父RDD的各个分片在集群中重新分片）

一个 Spark job 怎么执行的？

• Spark job 可以包含一系列对一组数据执行的操作。

不管 Spark job 的大小，Spark job 都需要 SparkContext 来执行。 在之前使用 REPL 的例子中，人们会注意到使用了一个名为 sc 的环境变量，这就是在 REPL 环境中如何访问一个 SparkContext。

👉 SparkContext 创建一个由不同 transformation 组成的 operator graph，一旦在某个 transformation 之上执行 action，这个 group 会被提交给 DAGScheduler。 根据 RDD 的性质或由窄变换（narrow transformation）或宽变换（wide transformation）（需要 shuffle 操作的变换）产生的结果。DAGScheduler 会创建 stage。
👉DAGScheduler 以这样的方式拆分 DAG：每个 stage 都由在通用洗牌边界（common shuffle boundaries，实在不知道怎么翻译合适）下的相同洗牌依赖（shuffle dependency）组成。 此外，stage 可以是洗牌映射（shuffle map）阶段，在这种情况下，其任务结果将是另一个阶段的输入；也可以是结果阶段（result stage），在这种情况下，其任务直接计算启动作业的 action，例如 count（）。
👉 然后 stage 被 DAGScheduler 作为任务集（TaskSets）提交给 TaskScheduler。 TaskScheduler 通过集群管理器（YARN，Mesos 和 Spark standalone）调度 TaskSet 并监视其执行情况。 如果任何任务失败，则重新运行，最后将结果发送到 DAGScheduler。 如果结果输出文件丢失，那么 DAGScheduler 将重新提交这些阶段给 TaskScheduler 以重新运行。
👉然后在满足资源和数据局部性约束的指定执行器（executor）（slave（worker）节点上运行的 JVM）上安排 task。 每个 executor 还可以分配多个 task。

一个spark程序的执行流程

1、启动：用户程序启动SparkContext，是程序的总入口，初始化过程中启动DAGScheduler作业调度和 TaskScheduler任务调度。
2、生成作业:DAGScheduler:根据shuffleDependency将作业划分为不同的stage,根据 RDD之间的依赖关系，宽依赖和窄依赖，划分原则就是遇见窄依赖就放进当前stage，遇到宽依赖则断开。(相当于shuffle是前后的stage分界线）每一个stage里面都会划分一个taskset,也就是数据集，而DAGSchedule的下一个任务就是将这个TaskSet传给TaskSchedule(在最后一个 stage划分结束,就会触发作业的提交)。
3、提交任务集: TaskScheduler:分配 Task到哪一个executor上去执行，SchedulerBackend配合TaskScheduler 完成具体任务的资源分配。
4、任务执行:Executor:实际任务的运行最终都 Execter 类来执行，对每个任务创建一个TaskRunner类，交给线程池去实现。

1. spark-submit 提交代码，执行 new SparkContext()，在 SparkContext 里构造 DAGScheduler 和 TaskScheduler。
2. TaskScheduler 会通过后台的一个进程，连接 Master，向 Master 注册 Application。
3. Master 接收到 Application 请求后，会使用相应的资源调度算法，在 Worker 上为这个 Application 启动多个 Executer。
4. Executor 启动后，会自己反向注册到 TaskScheduler 中。 所有 Executor 都注册到 Driver 上之后，SparkContext 结束初始化，接下来往下执行我们自己的代码。
5. 每执行到一个 Action，就会创建一个 Job。Job 会提交给 DAGScheduler。
6. DAGScheduler 会将 Job划分为多个 stage，然后每个 stage 创建一个 TaskSet。
7. TaskScheduler 会把每一个 TaskSet 里的 Task，提交到 Executor 上执行。
8. Executor 上有线程池，每接收到一个 Task，就用 TaskRunner 封装，然后从线程池里取出一个线程执行这个 task。(TaskRunner 将我们编写的代码，拷贝，反序列化，执行 Task，每个 Task 执行 RDD 里的一个 partition)

spark的stage是如何划分的

stage的划分依据就是看是否产生了shuflle(即宽依赖),遇到一个shuffle操作就划分为前后两个stage.

Spark的 RDD容错机制。

两个方法:利用“血缘（Lineage）容错”和检查点（checkpoint）机制。

1. “血缘”容错：利用依赖关系进行数据恢复，在容错机制中，如果一个节点死机了,而且运算窄依赖，则只要把丢失的父RDD分区重算即可，不依赖于其他节点。
   在窄依赖中，在子RDD 的分区丢失、重算父RDD分区时，父RDD相应分区计算所得到的数据都是子RDD分区的数据(一对一)，并不存在冗余计算。
   而宽依赖需要父RDD的所有分区都存在，重算花销大。在宽依赖情况下，丢失一个子RDD分区重算的每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的（其他子分区也会获得父亲重新计算的数据)，会有一部分数据相当于对应的是未丢失的子RDD分区中需要的数据，这样就会产生冗余计算开销，这也是宽依赖开销更大的原因。因此如果使用Checkpoint算子来做检查点。
2. 检查点机制：在宽依赖当中，如果 Lineage过长，重算开销会很大，通过设定检查点，在依赖关系中，对关系中间预算的结果进行数据冗余条份，所以数据恢复时，就可以l从检查点开始进行重新计算Lineage，减少开销。

checkpoint 检查点机制？

应用场景：当 spark 应用程序特别复杂，从初始的 RDD 开始到最后整个应用程序完成有很多的步骤，而且整个应用运行时间特别长，这种情况下就比较适合使用 checkpoint 功能。

原因：对于特别复杂的 Spark 应用，会出现某个反复使用的 RDD，即使之前持久化过但由于节点的故障导致数据丢失了，没有容错机制，所以需要重新计算一次数据。

Checkpoint 首先会调用 SparkContext 的 setCheckPointDIR() 方法，设置一个容错的文件系统的目录，比如说 HDFS；然后对 RDD 调用checkpoint()方法。之后在RDD所处的job运行结束之后，会启动一个单独的job，来将checkpoint 过的 RDD 数据写入之前设置的文件系统，进行高可用、容错的类持久化操作。

检查点机制是我们在 spark streaming 中用来保障容错性的主要机制，它可以使spark streaming阶段性的把应用数据存储到诸如 HDFS 等可靠存储系统中，以供恢复时使用。具体来说基于以下两个目的服务：

1. 控制发生失败时需要重算的状态数。Spark streaming 可以通过转化图的谱系图来重算状态，检查点机制则可以控制需要在转化图中回溯多远。
2. 提供驱动器程序容错。如果流计算应用中的驱动器程序崩溃了，你可以重启驱动器程序并让驱动器程序从检查点恢复，这样 spark streaming 就可以读取之前运行的程序处理数据的进度，并从那里继续。

Spark为什么快，Spark SQL 一定比 Hive 快吗

Spark SQL 比 Hadoop Hive 快，是有一定条件的，而且不是 Spark SQL 的引擎比 Hive 的引擎快，相反，Hive 的 HQL 引擎还比 Spark SQL 的引擎更快。其实，关键还是在于 Spark 本身快。

1. 消除了冗余的 HDFS 读写: Hadoop 每次 shuffle 操作后，必须写到磁盘，而 Spark 在 shuffle 后不一定落盘，可以 persist 到内存中，以便迭代时使用。如果操作复杂，很多的 shufle 操作，那么 Hadoop 的读写 IO 时间会大大增加，也是 Hive 更慢的主要原因了。
2. 消除了冗余的 MapReduce 阶段: Hadoop 的 shuffle 操作一定连着完整的 MapReduce 操作，冗余繁琐。而 Spark 基于 RDD 提供了丰富的算子操作，且 reduce 操作产生 shuffle 数据，可以缓存在内存中。
3. JVM 的优化: Hadoop 每次 MapReduce 操作，启动一个 Task 便会启动一次 JVM，基于进程的操作。而 Spark 每次 MapReduce 操作是基于线程的，只在启动 Executor 是启动一次 JVM，内存的 Task 操作是在线程复用的。每次启动 JVM 的时间可能就需要几秒甚至十几秒，那么当 Task 多了，这个时间 Hadoop 不知道比 Spark 慢了多少。

RDD 中 reduceBykey 与 groupByKey 哪个性能好，为什么

reduceByKey：reduceByKey 会在结果发送至 reducer 之前会对每个 mapper 在本地进行 merge，有点类似于在 MapReduce 中的 combiner。这样做的好处在于，在 map 端进行一次 reduce 之后，数据量会大幅度减小，从而减小传输，保证 reduce 端能够更快的进行结果计算。

groupByKey：groupByKey 会对每一个 RDD 中的 value 值进行聚合形成一个序列 (Iterator)，此操作发生在 reduce 端，所以势必会将所有的数据通过网络进行传输，造成不必要的浪费。同时如果数据量十分大，可能还会造成 OutOfMemoryError。

所以在进行大量数据的 reduce 操作时候建议使用 reduceByKey。不仅可以提高速度，还可以防止使用 groupByKey 造成的内存溢出问题。

RDD 持久化原理？

spark 非常重要的一个功能特性就是可以将 RDD 持久化在内存中。

调用 cache() 和 persist() 方法即可。cache() 和 persist() 的区别在于，cache() 是 persist() 的一种简化方式，cache() 的底层就是调用 persist() 的无参版本 persist(MEMORY_ONLY)，将数据持久化到内存中。

如果需要从内存中清除缓存，可以使用 unpersist() 方法。RDD 持久化是可以手动选择不同的策略的。在调用 persist() 时传入对应的 StorageLevel 即可。

checkpoint 和持久化机制的区别？

最主要的区别在于持久化只是将数据保存在 BlockManager 中，但是 RDD 的 lineage(血缘关系，依赖关系) 是不变的。但是 checkpoint 执行完之后，rdd 已经没有之前所谓的依赖 rdd 了，而只有一个强行为其设置的 checkpointRDD，checkpoint 之后 rdd 的 lineage 就改变了。

持久化的数据丢失的可能性更大，因为节点的故障会导致磁盘、内存的数据丢失。但是 checkpoint 的数据通常是保存在高可用的文件系统中，比如 HDFS 中，所以数据丢失可能性比较低

RDD懒加载是什么意思

Transformation 操作是延迟计算的，也就是说从一个RDD 转换生成另一个 RDD 的转换操作不是马上执行，需要等到有 Acion 操作的时候才会真正触发运算,这也就是懒加载.

spark 解决了 hadoop 的哪些问题(spark VS MR)？

1. MR：抽象层次低，需要使用手工代码来完成程序编写，使用上难以上手；

   Spark：Spark 采用 RDD 计算模型，简单容易上手。

2. MR：只提供 map 和 reduce 两个操作，表达能力欠缺；

   Spark：Spark 采用更加丰富的算子模型，包括 map、flatmap、groupbykey、reducebykey 等；

3. MR：一个 job 只能包含 map 和 reduce 两个阶段，复杂的任务需要包含很多个 job，这些 job 之间的管理以来需要开发者自己进行管理；

   Spark：Spark 中一个 job 可以包含多个转换操作，在调度时可以生成多个 stage，而且如果多个 map 操作的分区不变，是可以放在同一个 task 里面去执行；

4. MR：中间结果存放在 hdfs 中；

   Spark：Spark 的中间结果一般存在内存中，只有当内存不够了，才会存入本地磁盘，而不是 hdfs；

5. MR：只有等到所有的 map task 执行完毕后才能执行 reduce task；

   Spark：Spark 中分区相同的转换构成流水线在一个 task 中执行，分区不同的需要进行 shuffle 操作，被划分成不同的 stage 需要等待前面的 stage 执行完才能执行。

6. MR：只适合 batch 批处理，时延高，对于交互式处理和实时处理支持不够；

   Spark：Spark streaming 可以将流拆成时间间隔的 batch 进行处理，实时计算。

spark有哪几种join

Spark 中和 join 相关的算子有这几个：join、fullOuterJoin、leftOuterJoin、rightOuterJoin

• join
  join函数会输出两个RDD中key相同的所有项，并将它们的value联结起来，它联结的key要求在两个表中都存在，类似于SQL中的INNER JOIN。但它不满足交换律，a.join(b)与b.join(a)的结果不完全相同，值插入的顺序与调用关系有关。

• leftOuterJoin
  leftOuterJoin会保留对象的所有key，而用None填充在参数RDD other中缺失的值，因此调用顺序会使结果完全不同。如下面展示的结果，

• rightOuterJoin
  rightOuterJoin与leftOuterJoin基本一致，区别在于它的结果保留的是参数other这个RDD中所有的key。

• fullOuterJoin
  fullOuterJoin会保留两个RDD中所有的key，因此所有的值列都有可能出现缺失的情况，所有的值列都会转为Some对象。

hadoop 和 spark 的相同点和不同点？

Hadoop 底层使用 MapReduce 计算架构，只有 map 和 reduce 两种操作，表达能力比较欠缺，而且在 MR 过程中会重复的读写 hdfs，造成大量的磁盘 io 读写操作，所以适合高时延环境下批处理计算的应用；

Spark 是基于内存的分布式计算架构，提供更加丰富的数据集操作类型，主要分成转化操作和行动操作，包括 map、reduce、filter、flatmap、groupbykey、reducebykey、union 和 join 等，数据分析更加快速，所以适合低时延环境下计算的应用；

spark 与 hadoop 最大的区别在于迭代式计算模型。基于 mapreduce 框架的 Hadoop 主要分为 map 和 reduce 两个阶段，两个阶段完了就结束了，所以在一个 job 里面能做的处理很有限；spark 计算模型是基于内存的迭代式计算模型，可以分为 n 个阶段，根据用户编写的 RDD 算子和程序，在处理完一个阶段后可以继续往下处理很多个阶段，而不只是两个阶段。所以 spark 相较于 mapreduce，计算模型更加灵活，可以提供更强大的功能。

但是 spark 也有劣势，由于 spark 基于内存进行计算，虽然开发容易，但是真正面对大数据的时候，在没有进行调优的轻局昂下，可能会出现各种各样的问题，比如 OOM 内存溢出等情况，导致 spark 程序可能无法运行起来，而 mapreduce 虽然运行缓慢，但是至少可以慢慢运行完。

Hadoop/MapReduce 和 Spark 最适合的都是做离线型的数据分析，但 Hadoop 特别适合是单次分析的数据量 “很大” 的情景，而 Spark 则适用于数据量不是很大的情景。

1. 一般情况下，对于中小互联网和企业级的大数据应用而言，单次分析的数量都不会 “很大”，因此可以优先考虑使用 Spark。
2. 业务通常认为 Spark 更适用于机器学习之类的 “迭代式” 应用，80GB 的压缩数据（解压后超过 200GB），10 个节点的集群规模，跑类似 “sum+group-by” 的应用，MapReduce 花了 5 分钟，而 spark 只需要 2 分钟。

如何将spark-sql的Row转成Java对象？

Dataset转POJO

1. 将查询出的结果转为RDD
2. 将RDD创建为DataFrame，并传入schema参数
3. 调用as方法，将Dataset转为相应的POJO Dataset
4. 调用collectAsList()方法

SparkSession spark = CloudUtils.getSparkSession();// 查询原始数据
Dataset student = spark.sql("select * from `event`.`student`");
// 生成schema
List fields = new ArrayList<>();
fields.add(DataTypes.createStructField("id", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("major", DataTypes.StringType, true));
StructType schema = DataTypes.createStructType(fields);// 转换查询结果为POJO List
List students = spark.createDataFrame(student.toJavaRDD(), schema).as(Encoders.bean(Student.class)).collectAsList();
System.out.println(students);

Dataset中的日期类型为timestamp和java中的Date类型不兼容，和Timestamp类型相互兼容。
为了解决上述问题，我们可以先将Dataset转为JSON，然后将JSON转为POJO，代码

// 查出数据并转为json集合
List jsonList = spark.sql("select * from `event`.`user`").toJSON().collectAsList();
// 将json转为pojo，这里使用的是FastJSON        
List users = jsonList.stream().map(jsonString -> JSON.parseObject(jsonString, User.class)).collect(Collectors.toList());
System.out.println(users);

POJO转Dataset

// 获取users列表List users = createUsers();// 使用createDataFrame转为datasetDataset ds = spark.createDataFrame(users, User.class);// 将驼峰式列名改为下划线式列名，camelToUnderline方法网上搜索String[] columns = ds.columns();String[] newColumns = Arrays.stream(columns).map(column -> camelToUnderline(column)).toArray(String[]::new);// 转为新的df（重命名后的）ds.toDF(newColumns);ds.show();