在 Spark 中加入倾斜的数据集？

Question

我正在使用 Spark RDD 加入两个大型数据集。一个数据集非常倾斜，因此很少有执行程序任务需要很长时间才能完成工作。我该如何解决这种情况？

Answer 1

关于如何完成的相当不错的文章：https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

短版：

• 将随机元素添加到大型 RDD 并使用它创建新的连接键
• 使用explode/flatMap 将随机元素添加到小型RDD 以增加条目数并创建新的连接键
• 在新的连接键上加入 RDD，由于随机播种，现在可以更好地分布

Answer 2

假设您必须在 A.id=B.id 上连接两个表 A 和 B。让我们假设表 A 在 id=1 上存在偏差。

即从 A 中选择 A.id 在 A.id = B.id 上加入 B

解决倾斜连接问题有两种基本方法：

方法一：

将您的查询/数据集分成 2 个部分 - 一个仅包含倾斜数据，另一个包含非倾斜数据。 在上面的例子中。查询将变为 -

 1. select A.id from A join B on A.id = B.id where A.id <> 1;
 2. select A.id from A join B on A.id = B.id where A.id = 1 and B.id = 1;

第一个查询不会有任何偏差，因此 ResultStage 的所有任务将大致同时完成。

如果我们假设 B 只有几行 B.id = 1，那么它将适合内存。所以第二个查询将被转换为广播连接。这在 Hive 中也称为 Map-side join。

参考：https://cwiki.apache.org/confluence/display/Hive/Skewed+Join+Optimization

然后可以合并两个查询的部分结果以获得最终结果。

方法二：

上面 LeMuBei 也提到过，第二种方法尝试通过附加额外的列来随机化连接键。 步骤：

1. 在较大的表 (A) 中添加一列，例如 skewLeft，并为所有行填充 0 到 N-1 之间的随机数。

2. 在较小的表 (B) 中添加一列，例如 skewRight。将较小的表复制 N 次。因此，对于原始数据的每个副本，新 skewRight 列中的值将从 0 变化到 N-1。为此，您可以使用explode sql/dataset 运算符。

在 1 和 2 之后，连接 2 个数据集/表，连接条件更新为 -

                *A.id = B.id && A.skewLeft = B.skewRight*

参考：https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

Answer 3

根据您遇到的特定类型的偏差，可能有不同的解决方法。基本思路是：

• 修改您的连接列，或创建一个新的连接列，该列没有倾斜，但仍保留足够的信息来进行连接
• 在该非倾斜列上执行连接 - 生成的分区不会倾斜
• 连接后，您可以将连接列更新回您的首选格式，或者在创建新列时将其删除

如果倾斜的数据参与连接，李牧北的答案中引用的“Fighting the Skew In Spark”文章是一种很好的技术。就我而言，偏斜是由连接列中的大量空值引起的。空值不参与连接，但由于连接列上的 Spark 分区，连接后的分区非常倾斜，因为有一个包含所有空值的巨大分区。

我通过添加一个新列来解决它，该列将所有空值更改为分布良好的临时值，例如“NULL_VALUE_X”，其中 X 被替换为 1 到 10,000 之间的随机数，例如（在 Java 中）：

// Before the join, create a join column with well-distributed temporary values for null swids.  This column
// will be dropped after the join.  We need to do this so the post-join partitions will be well-distributed,
// and not have a giant partition with all null swids.
String swidWithDistributedNulls = "swid_with_distributed_nulls";
int numNullValues = 10000; // Just use a number that will always be bigger than number of partitions
Column swidWithDistributedNullsCol =
    when(csDataset.col(CS_COL_SWID).isNull(), functions.concat(
        functions.lit("NULL_SWID_"),
        functions.round(functions.rand().multiply(numNullValues)))
    )
    .otherwise(csDataset.col(CS_COL_SWID));
csDataset = csDataset.withColumn(swidWithDistributedNulls, swidWithDistributedNullsCol);

然后在这个新列上加入，然后在加入之后：

outputDataset.drop(swidWithDistributedNullsCol);

Answer 4

参考https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/ 下面是使用 Pyspark 数据帧 API 对抗 spark 偏差的代码

创建 2 个数据框：

from math import exp
from random import randint
from datetime import datetime

def count_elements(splitIndex, iterator):
    n = sum(1 for _ in iterator)
    yield (splitIndex, n)

def get_part_index(splitIndex, iterator):
    for it in iterator:
        yield (splitIndex, it)

num_parts = 18
# create the large skewed rdd
skewed_large_rdd = sc.parallelize(range(0,num_parts), num_parts).flatMap(lambda x: range(0, int(exp(x))))
skewed_large_rdd = skewed_large_rdd.mapPartitionsWithIndex(lambda ind, x: get_part_index(ind, x))

skewed_large_df = spark.createDataFrame(skewed_large_rdd,['x','y'])

small_rdd = sc.parallelize(range(0,num_parts), num_parts).map(lambda x: (x, x))

small_df = spark.createDataFrame(small_rdd,['a','b'])

大df将数据分成100个bin，小df复制100次

salt_bins = 100
from pyspark.sql import functions as F

skewed_transformed_df = skewed_large_df.withColumn('salt', (F.rand()*salt_bins).cast('int')).cache()

small_transformed_df = small_df.withColumn('replicate', F.array([F.lit(i) for i in range(salt_bins)]))

small_transformed_df = small_transformed_df.select('*', F.explode('replicate').alias('salt')).drop('replicate').cache()

终于避免了歪斜的连接

t0 = datetime.now()
result2 = skewed_transformed_df.join(small_transformed_df, (skewed_transformed_df['x'] == small_transformed_df['a']) & (skewed_transformed_df['salt'] == small_transformed_df['salt']) )
result2.count() 
print "The direct join takes %s"%(str(datetime.now() - t0))

Answer 5

Apache DataFu 有两种执行倾斜连接的方法，这些方法实现了先前答案中的一些建议。

joinSkewed 方法进行加盐（添加随机数列来拆分倾斜的值）。

broadcastJoinSkewed 方法适用于您可以将数据框划分为倾斜和常规部分，如moriarty007 的答案中的Approach 2 所述。

DataFu 中的这些方法对于使用 Spark 2.x 的项目很有用。如果您已经使用 Spark 3，则有 dedicated methods for doing skewed joins。

完全披露 - 我是 Apache DataFu 的成员。

Answer 6

您可以尝试将“偏斜”的 RDD 重新分区到更多分区，或者尝试增加 spark.sql.shuffle.partitions（默认为 200）。

在你的情况下，我会尝试将分区的数量设置为远高于执行者的数量。