Spark优化总结（一）——数据倾斜

1. 前言

• 在Spark应用开发过程中&＃xff0c;通常大多数性能问题是在数据倾斜点上。针对数据倾斜问题&＃xff0c;我们可以按运行状态分出以下几点&＃xff1a;
  • 数据源倾斜
  • 存储结果倾斜
  • 数据处理倾斜&＃xff08;问题最多的部分&＃xff09;

2. 数据源倾斜

• 说明

  • 使用Spark读取数据时&＃xff0c;通常我们的数据源也是分布式存储的&＃xff0c;这就需要Spark从数据源的各个节点分别读取数据。如果因为某些原因&＃xff0c;导致某份数据在存储集群的各个节点分布不均匀&＃xff0c;那么读取数据时需要的时间会等于最慢的节点
  • 例如&＃xff0c;数据分布在10个节点上&＃xff0c;其中一个节点的数据量是其他的节点的5倍

• 举例&＃xff1a;Hive分桶导致的数据倾斜

  • Hive设计表的时候可为表设定分桶&＃xff0c;选择一个良好的分布列能够使数据均匀的分布到集群中。
  • 如果分布列选择错误&＃xff0c;很可能导致某些节点数据极其多&＃xff0c;而其他节点又很少。那么&＃xff0c;使用Spark读取Hive表时&＃xff0c;就需要等待数据最多的节点执行完毕。

• 举例&＃xff1a;Kafka分区导致的数据倾斜

  • 生产到Kafka某个Topic的数据&＃xff0c;可以指定Partition、Key或者默认值&＃xff0c;用于决定数据会发送到Topic的那一个分区。
  • 如果生成的Partition、Key不对&＃xff0c;那么会导致Topic某些分区数据量远多于其他分区。那么&＃xff0c;使用SparkStream&＃xff08;或者StructuredStream&＃xff09;来接取Topic的数据时&＃xff0c;会导致Spark某个节点单次接入的数据过多。

3. 数据存储倾斜

• 说明

  • 通常利用Spark处理完数据后&＃xff0c;需要一个集群用于存储数据。一般来说&＃xff0c;HDFS、HIVE等集群不会存在存储性能问题。但是&＃xff0c;如果因为某些设计原因&＃xff0c;会导致待存储的数据其中大部分需要存入存储集群中的某个节点&＃xff08;或访问某个节点&＃xff09;&＃xff0c;显然Spark需要等待该节点存储完毕才能结束。

• 举例&＃xff1a;Hive分桶、Kafka分区

  • 和数据源倾斜同理

• 举例&＃xff1a;分布式MPP架构数据库

  • 一般来&＃xff0c;分布式数据库会有多个主节点用于JDBC/ODBC连接访问。如果用Spark存储时&＃xff0c;只设定一个主节点IP&＃xff0c;那么可能会导致分布式数据库该节点访问量激增&＃xff0c;从而导致存储效率下降。
  • 解决方案&＃xff1a;
    • Spark各节点分别随机选择&＃xff08;或者各自指定&＃xff09;数据库主节点IP
    • 使用数据库自带的数据导入工具&＃xff08;一般是批量导入&＃xff09;
  • 另外&＃xff0c;即便不是分布式数据库&＃xff0c;如果你的数据库只有一个节点可用&＃xff0c;同样会造成该问题

• 举例&＃xff1a;HBase的Rowkey设计错误

  • HBase的数据会根据rowkey进行自然排序&＃xff0c;排序相近的数据会存储在一个节点。如果同一时间向HBase某个表装入自然顺序相近的数据&＃xff0c;会导致Region热点问题。
  • 例如&＃xff0c;利用SparkStream实时采集数据装入HBase的某个表&＃xff0c;如果将该表的rowkey设计为以时间开头&＃xff0c;那么总是会入库自然顺序相近的数据&＃xff0c;导致热点问题&＃xff0c;最终入库性能下降。
  • 解决方案&＃xff1a;
    • 设计不以时间开头的rowkey
    • 如果每次查询是get&＃xff0c;能够具体知道rowkey&＃xff0c;那么可以将时间的后几位值移到前面来&＃xff0c;这样数据入库就随机了
    • 不需要实时入库时&＃xff0c;同时数据一般都需要在HIVE备份&＃xff0c;那么可以用Spark批量读取数据&＃xff0c;repartiton一下&＃xff0c;将数据顺序打乱&＃xff0c;然后入库HBase

4. 数据处理倾斜

4.1 过滤导致的数据不均

• 开发时&＃xff0c;在处理数据之前提前做好过滤&＃xff0c;是一个良好的习惯。但是&＃xff0c;过滤条件&＃xff08;例如where、filter&＃xff09;可能会导致部分分区的数据量较少&＃xff08;因为该分区符合过滤条件的数据少&＃xff09;&＃xff0c;而其他分区数据较多&＃xff0c;那么本次Stage执行的时间将取决于最慢的节点&＃xff08;数据最多&＃xff09;。

• 例如&＃xff1a;数据均匀的分布在10个分区&＃xff0c;每个节点分区约1000万条。你需要根据某个条件进行过滤数据&＃xff0c;然后再执行处理操作。当数据过滤后&＃xff0c;可能其中3个分区约有900万条符合条件&＃xff0c;其他7个分区约有200万条符合条件。接着继续执行后续逻辑&＃xff0c;显然本次Stage需要等数据最多的3个分区执行完才能结束&＃xff08;如果过滤后需要执行的处理逻辑较为花费时间&＃xff0c;例如每条数据还需要10秒&＃xff0c;那么效果尤为明显&＃xff09;。

• 解决方案&＃xff1a;

  • 使用repartion将数据再次分布均匀。需要注意的是repartion是shuffle操作&＃xff0c;网络传输需要消耗时间&＃xff0c;如果你的数据倾斜程度不大可以不做。也就是说你需要对比repartion的时间开销与数据倾斜导致的时间开销后&＃xff0c;再做决定。
  • 如果直接repartion也倾斜了&＃xff0c;那么你需要调用partitionBy()重写一个Patitioner&＃xff0c;选择合适的字段来做Shuffle。&＃xff08;没有合适的字段&＃xff0c;那就加随机前缀&＃xff09;
  • 关于coalesce&＃xff0c;在此几乎没有效果&＃xff0c;因为coalesce是合并各自节点本地的分区&＃xff0c;并没有解决数据倾斜的问题

• 简约示意图

  // 数据原状态| node1 | node2 | node3
count | 10050 | 10002 | 10035↓
// where/filter 根据条件过滤| node1 | node2 | node3
count | 5000 | 510 | 490↓
// 有的节点符合条件的多&＃xff0c;有的节点符合条件的少
// 如果继续进行map类型的处理&＃xff0c;node1将远远慢于node2、node3&＃xff0c;同时node2、node3还处于空闲状态&＃xff08;浪费&＃xff09;
// 如果每条数据的处理时间消耗比较大&＃xff0c;node1与node2、node3的差距将非常大↓
// 使用repartition&＃xff0c;重新分布| node1 | node2 | node3
count | 2014 | 1997 | 1989↓
// 继续进行map类型处理&＃xff0c;每个节点的计算资源都利用起来了
// ……


4.2 默认值、异常数据等导致的不均

• 开发中遇到数据倾斜&＃xff0c;可以优先排查该项。通常一个字段的处理中很可能会给一个默认值&＃xff08;例如0、null、空白字符等&＃xff09;或者一个异常值(例如&＃64;#&＃xffe5;%)&＃xff0c;这都是上游解析的问题导致的。
• 例如&＃xff0c;一个表共1亿条数据&＃xff0c;其中一列80%都是默认值字符串“01234”&＃xff0c;如果使用该列作为shuffle分区的key&＃xff0c;那么会导致数据倾斜。
• 建议提前统计好默认值或异常值的占比&＃xff08;例如抽样统计&＃xff09;&＃xff0c;如果不需要为该值的数据&＃xff0c;那么提前过滤掉。如果需要&＃xff0c;可以尝试加随机前缀&＃xff08;前提是能满足你的业务&＃xff09;。另外&＃xff0c;你可以换一种业务的实现方式&＃xff0c;想办法选择其他列作为shuffle的分区key。

4.3 join导致的数据不均

• 进行join时需要选择一个列&＃xff0c;如果该的列值倾斜那么join操作也会倾斜。
• 例如&＃xff1a;某批订单信息约1000万条&＃xff0c;其中有city字段&＃xff0c;其中A、B城市订单较多&＃xff08;各约300万&＃xff09;&＃xff0c;其他城市订单量较少&＃xff08;都少于50万&＃xff09;。此时需要将一份城市相关的数据与该数据关联&＃xff0c;如果选择city字段进行join&＃xff0c;会使各城市订单信息分别分布到各个分区节点上。显然A、B城市订单数据各自所在分区节点的数据量较大&＃xff0c;需要等待该分区处理完&＃xff0c;Stage才算结束。
• 解决方案&＃xff1a;
  • 使用广播进行 map-side join
  • 示例&＃xff1a;
    • Spark代码可读性与性能优化——示例五&＃xff08;HashJoin&＃xff09;

4.4 groupBy导致的数据不均

• 进行groupBy时需要选择一个列&＃xff0c;如果该的列值倾斜那么groupBy操作也会倾斜。
• 示例&＃xff1a;同join&＃xff0c;如果需要根据城市分组处理&＃xff0c;执行groupBy操作后&＃xff0c;因为A、B城市所在分区节点的数据量较大&＃xff0c;同样需要等待。
• 解决方案&＃xff1a;
  • 想办法使用reduceByKey、aggregateByKey、combineByKey替代
  • 示例&＃xff1a;
    • Spark代码可读性与性能优化——示例六&＃xff08;groupBy、reduceByKey、aggregateByKey&＃xff09;
    • Spark代码可读性与性能优化——示例八&＃xff08;一个业务逻辑&＃xff0c;多种解决方式&＃xff09;

4.5 key倾斜了&＃xff0c;但还必须要该key?

• 如果你的业务每天经常根据该key进行不同的数据处理的话&＃xff0c;那么可以尝试预先处理一次。
• 提前处理该key生成一份数据&＃xff0c;虽然第一次仍然会有倾斜问题&＃xff0c;速度较慢&＃xff0c;但是后面很多次的业务可以跟新新生成的数据进行处理&＃xff0c;不会再有该key倾斜的问题了。