Index

Have anice day :)
友链:

计算框架 Spark

目录

  1. 计算框架 Spark
    1. Spark
      1. Spark 中 job, stage, Task 分别是什么
      2. 什么是 RDD
        1. reduceByKey 和 groupByKey 的区别
        2. RDD 的创建方式
        3. RDD 的依赖关系
        4. 为什么要划分依赖关系
        5. RDD 的缓存持久化机制
      3. Spark 调优
        1. 修改序列化机制压缩数据量
        2. 根据作业环境, 提交任务时调整 spark-submit.sh 参数
        3. 提高并行度
        4. RDD 的重用和持久化
        5. 适当广播多次使用的变量
        6. 避免产生 Shuffle 可以减少网络 IO 和分区间传输消耗
        7. 使用 map-side 预先聚合 Shuffle
        8. 使用高性能算子
      4. SparkStreaming 在消费 Kafka 时下线, 如何保证重启后继续消费
      5. Spark 累加器
      6. Spark 的工作机制
      7. Spark 模块都有哪些
      8. DStream 和 DStreamGraph 的区别

Spark

Spark 中 job, stage, Task 分别是什么

什么是 RDD

RDD 就是弹性分布式数据集 Resilient Distributed Datasets, Spark 的数据抽象概念, 是一种不可变分布式对象集合; 每个 RDD 都被分为多个分区, 运行在集群不同节点上, 拥有多种不同的 RDD 算子.

reduceByKey 和 groupByKey 的区别

RDD 的创建方式

使用 makeRDD 通过集合创建, 由本地核数决定分区数量.

使用外部数据源 HDFS 的时候, 由 Block 数量决定分区数, 最低是两个.

由另一个 RDD 得出的结果创建, 即转换时创建, 会根据父 RDD 的 reduceTask 数量决定分区.

RDD 的依赖关系

  1. 宽依赖: 多个子 RDD 的分区依赖同一个父 RDD 的 Partition.
  2. 窄依赖: 父 RDD 的 Partition 至多被子 RDD 的一个 Partition 使用.

为什么要划分依赖关系

有效区分算子间的关系, 有利于 DAG 图形生和运行时依赖监察;

如果将依赖关系混淆, 将会导致运行效率的分配不均.

RDD 的缓存持久化机制

主要通过 cache, persist, checkpoints 实现.

  1. cache 与 persist:
    • cache 默认将数据存储在内存中, 靠 persist 实现,
    • persist 定义了多种数据储存策略, 如磁盘内存多副本. 数据存储在内存中会产生内存溢出, 断点失效等问题, 不能保证数据准确和安全.
    • 不改变 RDD 的依赖关系, 程序执行完成后回收缓存.
    • 触发 cache 和 persist 需要通过 action.
    • 不会开启其他 job, 一个 action 一个 job 原则.
  2. checkpoints:
    • 将数据存储在 HDFS 上实现持久化.
    • 触发 checkpoints 需要通过 action, 会开启新的 job.
    • 会改变 RDD 依赖关系, 数据丢失后无法通过依赖关系恢复数据; 程序执行结束后不会回收.

Spark 调优

修改序列化机制压缩数据量

通过 Kryo 调整序列化性能.

根据作业环境, 提交任务时调整 spark-submit.sh 参数

# 配置 Executor 数量
 --num-Executors 3
# 配置 driver 内存, 单位 gb
 --driver-memory ng
# 配置 Executor 的内存大小, 单位 gb
 --Executor-memory n
# 配置 Executor cpu 核心使用个数, 单位 个
 --Executor-cores n

提高并行度

RDD 的重用和持久化

一个副本时服务宕机了就得重新计算, 持久化每个数据单元, 并多做冗余互备来容错. 适合大内存场景. StorageLevel.MEMORY_ONLY_2.

适当广播多次使用的变量

通过 sparkContext 的 broadcast 方法把数据转换成广播变量:

val broadcastArray:Broadcast[Array[Int]]=sc.broadcast(Array(1,2,3,4,5,6))
        sc.broadcast(Array(1,2,3,4,5,6))

Executor 的 BlockManager 就可以拉取该广播变量的副本获取具体的数据, 获取广播变量中的值可以通过调用其 value 方法:

val array:Array[Int]=broadcastArray.value

注意:

  1. RDD 没法广播, 但是 RDD 结果可以广播.
  2. Spark 中因为算子的真正逻辑是发送给 Executor 运行的, Executor 需要引用外部变量时, 可以使用广播变量.
  3. 广播变量只能在 Driver 上定义和修改变量值.
  4. 若 Executor 调用了 Driver 的变量, 不使用广播时有几个 Task 就要创建几个 Driver 的变量副本. 用了广播后只创建一个副本.

避免产生 Shuffle 可以减少网络 IO 和分区间传输消耗

// join 会产生 Shuffle
// 两个 RDD 相同的 key 需要通过网络 PULL 到一个节点上, 由一个 Task 执行 join
// val rdd3 = rdd1.join(rdd2)
// 广播变量和 Map 搭配的 join 操作不会产生 Shuffle
val rdd2Data=rdd2.collect()
        val rdd2DataBroadcast=sc.broadcast(rdd2Data)
// 每个 Executor 的内存都会缓存一份 rdd2 的全量数据
// 需要结合实际硬件性能考虑广播
// 在 rdd1.map() 中, 可以从 rdd2DataBroadcast 中获取 rdd2 的所有数据
// 若遍历结果表示 rdd2 中某条数据的 key 与 rdd1 的当前数据 key 相同, 就可以 join
// 再根据业务将 rdd1 与 rdd2 可连接的数据拼接成 String 或 Tuple
        val rdd3=rdd1.map(rdd2DataBroadcast)

使用 map-side 预先聚合 Shuffle

使用高性能算子

  1. 使用 mapPartitions() 替代 map()
    • mapPartitions() 类的算子每次调用会处理 Partition 中所有的数据, 相比单条处理效率高些, 但会占用大量内存.
  2. 使用 foreachPartition() 替代 foreach()
    • 原理同上.
  3. 使用 Filter() 之后进行 coalesce()
    • 对一个 RDD 过滤掉较多数据后可以手动减少多余的 Partition 数量.
  4. 使用 repartitionAndSortWithinPartitions() 替代 repartition() 与 sort() 类操作
    • Spark 官方推荐在 repartition() 后的排序场景下使用这个名字很长的算子, 该算子可以并行执行重新分区的 shuffle 和排序.
  5. 使用 fastutil() 优化数据格式
    • fastutil 是 java 标准集合框架的扩展库, 提供了特殊的 map, set, list, queue; fastutil 能减少内存开销. 除了一些使用外部变量的场景外, 对付某种较大集合时也可以用 fastutil 改写外部变量.

fastutil 需要用 maven 手动引入依赖.

SparkStreaming 在消费 Kafka 时下线, 如何保证重启后继续消费

Spark 累加器

累加器相当于统筹大变量, 常用于计数统计工作. 累加器通常被视为 RDD 的 map().filter() 操作的副产物.

Spark 的工作机制

  1. Client 提交 job 后, Driver 运行 main 方法并创建 SparkContext
  2. 执行 RDD 算子后形成 DAG 图, 再移交给 DAGScheduler 处理.
  3. DAGScheduler 按照 RDD 的依赖关系划分 stage, 输入 Task, Scheduler 去划分 Task, .
  4. set 分发到各个节点的 Executor, 并以多线程的方式执行 Task, 一个线程一个 Task, 任务结束后根据类型返回结果.

Spark 模块都有哪些

DStream 和 DStreamGraph 的区别