大家好,我是梦想家Alex ~ 今天为大家带来大数据开发面试中,关于 Spark 的 28 个高频考点 。
适当增加 spark standby master
编写 shell 脚本,定期检测 master 状态,出现宕机后对 master 进行重启操作
Spark streaming 是 spark core API 的一种扩展,可以用于进行大规模、高吞吐量、容错的实时数据流的处理。
它支持从多种数据源读取数据,比如 Kafka、Flume、Twitter 和 TCP Socket,并且能够使用算子比如 map、reduce、join 和 window 等来处理数据,处理后的数据可以保存到文件系统、数据库等存储中。
Spark streaming 内部的基本工作原理是:接受实时输入数据流,然后将数据拆分成 batch ,比如每收集一秒的数据封装成一个 batch,然后将每个 batch 交给 spark 的计算引擎进行处理,最后会生产处一个结果数据流,其中的数据也是一个一个的 batch 组成的 。
master:管理集群和节点,不参与计算。
worker:计算节点,进程本身不参与计算, 和 master 汇报。Driver:运行程序的 main 方法,创建 spark context 对象。
spark context:控制整个 application 的生命周期,包括 dagsheduler 和 task scheduler 等组件。
client:用户提交程序的入口。
用户在 client 端提交作业后,会由 Driver 运行 main 方法并创建 spark context 上下文。执行 add 算子,形成 dag 图输入 dagscheduler,按照 add 之间的依赖关系划分 stage 输入 task scheduler。task scheduler 会将 stage 划分为 task set 分发到各个节点的 executor 中执行。
Master 实际上可以配置两个,Spark 原生的 standalone 模式是支持 Master 主备切换的。当 Active Master 节点挂掉以后,可以将 Standby Master 切换为 Active Master。
Spark Master 主备切换可以基于两种机制,一种是基于文件系统的,一种是基于 ZooKeeper 的。
基于文件系统的主备切换机制,需要在 Active Master 挂掉之后手动切换到 Standby Master 上;而基于 Zookeeper 的主备切换机制,可以实现自动切换 Master 。
1)本地模式 Spark 不一定非要跑在 hadoop 集群,可以在本地,起多个线程的方式来指定。将 Spark 应用以多线程的方式直接运行在本地,一般都是为了方便调试 。
本地模式分三类
2)standalone 模式
分布式部署集群,自带完整的服务,资源管理和任务监控是 Spark 自己监控,这个模式也是其他模式的基础 。
3)Spark on yarn 模式
分布式部署集群,资源和任务监控交给 yarn 管理,但是目前仅支持粗粒度资源分配方式,包含 cluster 和 client 运行模式,cluster 适合生产,driver 运行在集群子节点,具有容错功能,client 适合调试,dirver 运行在客户端。
4)Spark On Mesos 模式
官方推荐这种模式(当然,原因之一是血缘关系)。正是由于 Spark 开发之初就考虑到支持 Mesos,因此,目前而言,Spark 运行在 Mesos 上会比运行在 YARN 上更加灵活,更加自然。用户可选择两种调度模式之一运行自己的应用程序:
1)基于内存计算,减少低效的磁盘交互;
2)高效的调度算法,基于 DAG;
3)容错机制 Linage,精华部分就是 DAG 和 Lingae
1)从 high-level 的角度来看,两者并没有大的差别。 都是将 mapper (Spark 里是 ShuffleMapTask)的输出进行 partition,不同的 partition 送到不同的 reducer(Spark 里 reducer 可能是下一个 stage 里的 ShuffleMapTask,也可能是 ResultTask)。Reducer 以内存作缓冲区,边 shuffle 边 aggregate 数据,等到数据 aggregate 好以后进行 reduce() (Spark 里可能是后续的一系列操作)。
2)从 low-level 的角度来看,两者差别不小。 Hadoop MapReduce 是 sort-based,进入 combine() 和 reduce() 的 records 必须先 sort。这样的好处在于 combine/reduce() 可以处理大规模的数据,因为其输入数据可以通过外排得到(mapper 对每段数据先做排序,reducer 的 shuffle 对排好序的每段数据做归并)。目前的 Spark 默认选择的是 hash-based,通常使用HashMap 来对 shuffle 来的数据进行 aggregate,不会对数据进行提前排序。如果用户需要经过排序的数据,那么需要自己调用类似 sortByKey() 的操作;如果你是 Spark 1.1 的用户,可以将 spark.shuffle.manager 设置为 sort,则会对数据进行排序。在 Spark 1.2 中,sort 将作为默认的 Shuffle 实现。
3)从实现角度来看,两者也有不少差别。 Hadoop MapReduce 将处理流程划分出明显的几个阶段:map(), spill, merge, shuffle, sort, reduce() 等。每个阶段各司其职,可以按照过程式的编程思想来逐一实现每个阶段的功能。在 Spark 中,没有这样功能明确的阶段,只有不同的 stage 和一系列的 transformation(),所以 spill, merge, aggregate 等操作需要蕴含在 transformation() 中。如果我们将 map 端划分数据、持久化数据的过程称为 shuffle write,而将reducer 读入数据、aggregate 数据的过程称为 shuffle read。那么在 Spark 中,问题就变为怎么在 job 的逻辑或者物理执行图中加入 shuffle write 和 shuffle read 的处理逻辑?以及两个处理逻辑应该怎么高效实现?Shuffle write 由于不要求数据有序,shuffle write 的任务很简单:将数据 partition 好,并持久化。之所以要持久化,一方面是要减少内存存储空间压力,另一方面也是为了 fault-tolerance。
① 构建 Application 的运行环境,Driver 创建一个 SparkContext
② SparkContext 向资源管理器(Standalone、Mesos、Yarn)申请Executor 资源,资源管理器启动 StandaloneExecutorbackend(Executor)
③ Executor 向 SparkContext 申请 Task
④ SparkContext 将应用程序分发给 Executor
⑤ SparkContext 就建成 DAG 图,DAGScheduler 将 DAG 图解析成 Stage,每个 Stage 有多个 task,形成 taskset 发送给 task Scheduler,由task Scheduler 将 Task 发送给 Executor 运行
⑥ Task 在 Executor 上运行,运行完释放所有资源
spark 调优比较复杂,但是大体可以分为三个方面来进行:
1)平台层面的调优:防止不必要的 jar 包分发,提高数据的本地性,选择高效的存储格式如 parquet
2)应用程序层面的调优:过滤操作符的优化降低过多小任务,降低单条记录的资源开销,处理数据倾斜,复用 RDD 进行缓存,作业并行化执行等等
3)JVM 层面的调优:设置合适的资源量,设置合理的 JVM,启用高效的序列化方法如 kyro,增大 off head 内存等等
具体的 task 运行在那他机器上,dag 划分 stage 的时候确定的
1)自动的进行内存和磁盘的存储切换; 2)基于 Lineage 的高效容错;
3)task 如果失败会自动进行特定次数的重试;
4)stage 如果失败会自动进行特定次数的重试,而且只会计算失败的分片;
5)checkpoint 和 persist,数据计算之后持久化缓存;
6)数据调度弹性,DAG TASK 调度和资源无关;
7)数据分片的高度弹性。
1)不支持细粒度的写和更新操作(如网络爬虫),spark 写数据是粗粒度的。所谓粗粒度,就是批量写入数据,为了提高效率。但是读数据是细粒度的也就是说可以一条条的读 。
2)不支持增量迭代计算,但是 Flink 支持
可以从下面三点去展开
Spark 中的数据本地性有三种:
通常读取数据 PROCESS_LOCAL>NODE_LOCAL>ANY,尽量使数据以 PROCESS_LOCAL 或 NODE_LOCAL 方式读取。其中 PROCESS_LOCAL 还和 cache 有关,如果 RDD 经常用的话将该 RDD cache 到内存中,注意,由于cache 是 lazy 的,所以必须通过一个 action 的触发,才能真正的将该 RDD cache 到内存中。
为什么要进行持久化?
spark 所有复杂一点的算法都会有 persist 身影,spark 默认数据放在内存,spark 很多内容都是放在内存的,非常适合高速迭代,1000 个步骤只有第一个输入数据,中间不产生临时数据,但分布式系统风险很高,所以容易出错,就要容错,rdd 出错或者分片可以根据血统算出来,如果没有对父 rdd 进行 persist 或者 cache 的话,就需要重头做。 正常情况下,以下场景会使用 persist:
join 其实常见的就分为两类: map-side join 和 reduce-side join
。当大表和小表 join 时,用 map-side join 能显著提高效率。将多份数据进行关联是数据处理过程中非常普遍的用法,不过在分布式计算系统中,这个问题往往会变的非常麻烦,因为框架提供的 join 操作一般会将所有数据根据 key 发送到所有的 reduce 分区中去,也就是 shuffle 的过程。造成大量的网络以及磁盘 IO 消耗,运行效率极其低下,这个过程一般被称为 reduce-side-join。如果其中有张表较小的话,我们则可以自己实现在 map 端实现数据关联,跳过大量数据进行 shuffle 的过程,运行时间得到大量缩短,根据不同数据可能会有几倍到数十倍的性能提升 。
需要注意一个细节: 注意:如果 ResourceManager 第一次没有能够满足 driver(ApplicationMaster)的资源请求 ,后续发现有空闲的资源,会主动向 driver(ApplicationMaster)发送可用资源的元数据信息以提供更多的资源用于当前程序的运行。
不一定,当数据规模小,Hash shuffle 快于 Sorted Shuffle ;数据规模大的时候,sorted Shuffle 会比 Hash shuffle 快很多 。因为数量大的有很多小文件,不均匀,甚至出现数据倾斜,消耗内存大,1.x 之前 spark 使用 hash,适合处理中小规模,1.x 之后,增加了 Sorted shuffle,Spark 更能胜任大规模处理了 。
spark.storage.memoryFraction
参数,使得更多的持久化数据保存在内存中,提高数据的读取性能,如果 shuffle 的操作比较多,有很多的数据读写操作到 JVM 中,那么应该调小一点,节约出更多的内存给 JVM,避免过多的 JVM gc 发生。在 WEB UI 中观察如果发现 gc 时间很长,可以设置 spark.storage.memoryFraction
更小一点。本篇文章总结了在大数据开发面试过程中关于 Spark 的核心知识点 。大家不用去硬记,但至少被问到了需要有印象且能按照正确的思路去回答 。好了,本篇文章就到这里,下周我将为大家带来更加硬核的内容,欢迎持续关注,我们下期不见不散!