1. 性能优化

代码细节

Spark 是惰性求值的，只有当必要的时候才会读入数据，当然一个数据可能读入多次。这样可以方便语法分析器优化执行的方案。

PySpark 中，当用类包装 Function 的时候，不要传入 self.xxx，而是需要把 self.xxx 转成一个局部变量然后传入，避免把整个类都传入。

用 collect 方法的时候，所有的结果都会聚集到一台机子上，可能会导致内存不足，如果只是需要把节点上的数据发送到某处可以直接用 forEach 操作发送，而不需要把 RDD 聚集到本地。当然调试的时候可以用 take 方法进行采样，这个方法会尽量少的从节点取出数据。

1.1 内存优化

Spark 的内存分为三块区域，可以通过调整这几个区域占的比重调整性能

RDD 存储，调用 cache 或者 persist 的时候会用到
数据混洗和聚合的缓存区
用户代码使用

如果想要不重新计算，而是重复使用同一个 RDD，可以使用 persist 或者 cache 方法来持久化。如果一个保存持久化数据节点出了故障，则会重新计算 RDD，默认持久化级别是保存在 JVM 堆中，可以通过参数调整保存在硬盘或者堆中。

1.2 并行度优化

并行度会影响性能，并行度过大会导致切换开销过大，并行度过小会导致资源闲置。

Spark 会针对 RDD 自动推算出并行度，混洗后会继承父 RDD 的并行度。

比如说对于一个程序，filter 操作过滤掉了绝大部分数据，则可以通过 coalesce() 操作来调整分区数目，进而调整并行度，这个操作比 repartition 更高效一些。

1.3 序列化格式优化

利用 Kryo 序列化库是你提升 Spark 性能要做的第一件事

https://www.jianshu.com/p/8ccd701490cf

1.4 分区优化

通过控制键值对来把频繁一起被访问的 RDD 放在同一个节点上可大大减少应用通信开销，明显提升性能。

数据分区

比如说在 join 操作中：

将两个 RDD 的所有 key 的 hash 值算出来
通过 hash 把 RDD 分片到不同节点上
进行 join 操作

如果会出现多次对同一个 RDD join 的情况，可以这样优化

分区，在节点上保存 RDD 数据
只需要对每次新出现的 RDD 进行求 hash 值，分配到不同节点上
执行 join 操作

所以量比较大的静态数据最好能分区后持久化存储

不是所有的操作都可以从分区上受益，比如说 map 操作可能会改变 key 的值，那在原来的分区上就可能找不到这个 key 了，Spark 不会判断是否改变了 key，而是选择每次都重新计算 key，所以 map 操作不能从分区上受益。如果确定不会改变到 key 则可以用 mapValues 代替。

可以从分区获益的操作：

cogroup()
groupWith()
join()
left right join
groupByKey
reduceByKey
combineByKey
partitionBy
sort
mapValues

分区操作优化

对于一些代价比较大的操作，比如说打开数据库连接，不可能在 forEach 中每一个循环都打开一遍，则就需要用到分区操作，也就是说每一个分区都打开一个数据库连接即可，可以大大减少打开数据库连接的时间（当然使用广播线程池会更好点）

1.5 全局共享优化

累加器

累加器相当于全局变量，但是如果在 Spark 中直接使用全局变量的话，相当于会在每一个节点上复制相同的一份全局变量，全局变量就会失效。

text = 'a b c a a c b'
rdd = sc.parallelize(text.split(' '))

def invalid_sign(sign):
    global validSignCount, invalidSignCount
    if sign == '':
        invalidSignCount += 1
        return False
    else:
        validSignCount += 1
        return True

print(rdd.filter(invalid_sign).collect())
print(validSignCount.value)

广播变量

当在许多个节点上需要共享同一份变量的时候，则需要用到广播变量，广播变量只读不写。

public_value = sc.broadcast({'a': 1, 'b': 2, 'c': 3})


def map_value(char, mapper_broadcast):
    return mapper_broadcast.value[char]


print(rdd.map(lambda x: (x, map_value(x, public_value))).reduceByKey(lambda x, y: x + y))

本文由鸡米创作，采用知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Nov 13,2019

当我谈 Spark 时我谈些什么