规范化的SPARK RDD分区，使用减法或重新分区

Question

使用Spark 2.4.0。我的生产数据非常歪斜，因此其中一项任务的时间是其他任务的7倍。我尝试了不同的策略来规范数据，以便所有执行者都能平等工作-

1. spark.default。并行性
2. reduceByKey（numPartitions）
3. 重新分区（numPartitions）

我的期望是这三个选项应该均匀分区，但是在Spark Local/Standalone上使用一些虚拟的非生产数据表明，选项1、2比3更正常。

数据如下：

account}date}ccy}amount
A1}2020/01/20}USD}100.12
A2}2010/01/20}SGD}200.24
A2}2010/01/20}USD}300.36
A1}2020/01/20}USD}400.12

预期结果应该是[A1-USD，500.24]，[A2-SGD，200.24]，[A2-USD，300.36]理想情况下，这些应该划分为3个不同的分区。

javaRDDWithoutHeader
.mapToPair((PairFunction<Balance, String, Integer>) balance -> new Tuple2<>(balance.getAccount() + balance.getCcy(), 1))        
    .mapToPair(new MyPairFunction())
   .reduceByKey(new ReductionFunction())

检查分区的代码

     System.out.println("b4 = " +pairRDD.getNumPartitions());
     System.out.println(pairRDD.glom().collect());
     JavaPairRDD<DummyString, BigDecimal> newPairRDD = pairRDD.repartition(3);
     System.out.println("Number of partitions = " +newPairRDD.getNumPartitions());
     System.out.println(newPairRDD.glom().collect());

>

选项1：什么都不做

选项2：将spark.default.parallelism设置为3

选项3：使用Num分区=3的减除ByKey

选项4：重新分区（3）

对于选项1分区数=2[[[（DummyString{帐户='A2'， ccy='SGD'}，200.24），（DumMyString{帐户='A2'， ccy='USD'}，300.36）]，[（DumMyString{帐户='A1'， ccy='USD'}，500.24）]]

对于备选方案2

分区数 = 3 [ [（DummyString{account='A1'， ccy='USD'}，500.24）]， [（DummyString{account='A2'， ccy='USD'}，300.36）]， [（DummyString{account='A2'， ccy='SGD'}，200.24）]]

对于选项 3 分区数 = 3 [ [（DummyString{account='A1'， ccy='USD'}，500.24）]， [（DummyString{account='A2'， ccy='USD'}，300.36）]， [（DummyString{account='A2'， ccy='SGD'}，200.24）]

对于选项 4 分区数 = 3 [[]， [（DummyString{ account='A2'， ccy='SGD'}，200.24）]， [（DummyString{ account='A2'， ccy='USD'}，300.36）， （DummyString{ account='A1'， ccy='USD'}，500.24）]]

结论：选项 2（spark.default.parallelism） 和 3（reduceByKey（numPartitions） 规范化比选项 4（重新分区）好得多 相当确定的结果，从未看到选项4规范化为 3 个分区。

问题:

1. 是否比重新分区或
2. 这只是因为样本数据集太小了吗？或者
3. 当我们通过YARN集群提交时，这种行为会有所不同吗

Answer 1

我认为这个问题中有一些东西贯穿始终，因此更难回答。

首先，存在与静态数据相关的分区和并行性，因此在读取时也是如此；在不让海洋重新沸腾的情况下，这里有一个极好的SO答案来解决这个问题：当spark的主内存无法容纳文件时，spark如何读取一个大文件（PB）。在任何情况下，都没有散列或任何正在进行的事情，只是“原样”。

此外，与DFs相比，rdd没有得到很好的优化。

Spark中的各种操作会在一个动作被调用后导致洗牌:

• reduceByKey将使用散列进行最终聚合和本地分区聚合，这将导致更少的混洗
• 重新分区，使用随机性
• partitionBy（new HashPartitioner（n））等，您没有提到
• reduceByKey（聚合函数，N个分区），奇怪的是，它比重新分区更有效

您的后一条评论通常暗示了数据的偏差。对于reduceByKey，太多条目哈希到同一个“桶”/分区。缓解方法：

>

一般来说，尝试预先使用大量分区（读入时） - 但我在这里看不到您的转换和方法，因此我们将其作为一般建议。

一般来说，尝试使用合适的哈希在前面（读入时）使用更多的分区，但我看不到这里的转换和方法，所以我们将此作为一般建议。

或者在某些情况下，通过添加后缀来“加盐”密钥，然后 reduceByKey 并再次将 reduceByKey “取消盐化”以获取原始密钥。取决于所花费的额外时间与保持原样或执行其他选项。

repartition（n） 应用随机排序，因此您洗牌，然后需要再次洗牌。不必要的海事组织。正如另一篇文章所示（请参阅对您的问题的评论），它看起来像是不必要的工作，但这些都是旧式的RDD。

顺便说一句，使用数据帧更容易做到。

由于我们不了解您的完整编码，希望对您有所帮助。