偏斜数据的自定义分区器答案

【问题标题】：Custom Partitioner for skewed data偏斜数据的自定义分区器
【发布时间】：2019-05-25 11:09:47
【问题描述】：

我有一些用例的全球数据。每个国家/地区包含 3 到 5 种产品，每小时为每个用户收集。我想做引导来计算每个国家/地区每小时每个产品的一些平均比率和其他比率。

input.rdd.map(
      row => (
        (country, product, hour),
        (country, product, hour, user, rating)
        )
    )
val groups = keyGroup.groupByKey()
val output = groups.flatMapValues(x => bootstrap(x)).toDF

问题是一些国家的数据非常大，这导致整个过程需要数小时但仍未完成。我试着得到大致的尺寸：

Partition:count ->Countries

0: 2044816 -> India,Turkey

1: 1466790 -> Turkey,India

2: 783772 -> India,Mexico,Japan,South Korea

3: 431538 -> Japan,Mexico,South Korea,India,Indonesia,Turkey,Brazil,Russian Federation

4: 319824 -> South Korea,Brazil,Russian Federation,India,Mexico,United States of America,Turkey,Japan,Bangladesh

5: 268698 -> Bangladesh,Nigeria,Russian Federation,United States of America

6: 264709 -> Russian Federation,United States of America,Germany,Bangladesh,Nigeria,South Africa

7: 227612 -> South Africa,United States of America,Russian Federation,Brazil,South Korea,Germany
...
...
167: 58 -> Mexico,Chile,Uganda,Thailand,Ivory Coast,Antigua and Barbuda,Palau,Luxembourg,United States of America,British Virgin Islands,Iceland,Andorra,Samoa,Vanuatu,Botswana,Saint Lucia,Kiribati,Greenland

168: 69 -> Greenland,Iceland,Chile,Zambia,Estonia,Vanuatu,Cyprus,Malta,Saudi Arabia,Japan,Uruguay,Qatar,United States of America,Luxembourg,Peru,Belize,Papua New Guinea,Samoa,South Sudan

169: 61 -> Myanmar,Belize,Chile,Somalia,Bhutan,Luxembourg,Liberia,Norway,United Kingdom,Burkina Faso,Lithuania,Macedonia,Belgium,Vanuatu,Burundi,DR Congo,Montenegro,Central African Republic,Bosnia and Herzegovina

170: 36 -> Mauritania,Sierra Leone,Hungary,Zambia,Somalia,Federated States of Micronesia,Serbia,Liberia,Nepal,Chile,Israel,Ukraine,Montenegro,Yemen,Croatia,Central African Republic,Armenia,Andorra,United Arab Emirates,Mauritius,Albania,Lebanon,Macedonia

171: 25 -> Spain,Comoros,Libya,Peru,Latvia,Montenegro,Egypt,Malaysia,Central African Republic,Faroe Islands,Tanzania,Palau,Chad,Guatemala,Kiribati,Burundi,Luxembourg,Equatorial Guinea,Barbados,Belgium

172: 14 -> Vietnam,Tanzania,Hungary,Egypt,Comoros,Equatorial Guinea,Guinea-Bissau,Moldova,Macedonia,Guyana,Federated States of Micronesia,New Zealand,Chad

可以看出，数据划分不均，有173个分区。数据约为 6 GB，其中包含一周的数据。如果我尝试通过执行 repartition of 1000 来运行单个国家/地区，它可以工作，但在一起它就行不通。

我正在考虑编写一个自定义分区器，但我不知道我应该如何打破更大国家/地区的数据。如果有人可以帮助我，那就太好了。

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

在您开始编写自己的自定义分区程序之前，您可以尝试以下操作：由于您已经知道哪些国家/地区存在倾斜数据，您可以为所述国家/地区创建一个组合键，其中随机附加数字在某个范围内（更广泛的范围以获得更多倾斜数据）。您可以在所述键上聚合，然后删除复合键并进一步聚合。

df.withColumn("composite_key", 
    when(isSkewDataCountryUDF(col("country")), concat(col("country"), randomNumberSuffix())
    .otherwise(col("country")))
.groupBy("composite_key")
.count
.drop("composite_key")
.groupBy("country")
.count

同时尝试设置更高的spark.default.parallelism 和spark.sql.shuffle.partitions

【讨论】：