分区列倾斜时更有效地写入分区拼花答案

【问题标题】：More efficiently writing partitioned parquet when partitioning column is skewed分区列倾斜时更有效地写入分区拼花
【发布时间】：2019-07-10 18:03:48
【问题描述】：

我正在用分区拼花写一个大表（大约 1.2b 行），我使用状态（如美国州）作为分区键。问题是存在大量空状态值。此表通常按状态查询，因此具有空状态的大分区不是问题，但我在更有效地生成表时遇到了麻烦。

我尝试使用非空状态创建表，然后插入空值，但据我所知，所有空值仍然只是放在一个大分区中，因此发送给一个工作人员。

如果有一种方法可以插入特定的分区，那就太好了。就像我的示例一样，写入非空状态，然后将剩余的记录插入到 state=null 或 hive_default_partition 中，这种方式仍然可以跨集群并行化。

【问题讨论】：

标签： sql apache-spark partitioning skew

【解决方案1】：

尝试使用自动分区写入非空数据，然后将空数据重新分区并单独写入，例如：

df.where($”state”.isNotNull).write.partitionBy($”state”).parquet(“my_output_dir”)
df.where($”state”.isNull).repartition(100).write.parquet(“my_output_dir/state=__HIVE_DEFAULT_PARTITION__”)

使用 SQL API，您可以使用重新分区提示（在 Spark 2.4 中引入）来完成相同的操作：

spark-sql> describe skew_test;  
id  bigint  NULL
dt  date    NULL
state   string  NULL
# Partition Information     
# col_name  data_type   comment
state   string  NULL
Time taken: 0.035 seconds, Fetched 6 row(s)
spark-sql> CREATE TABLE `skew_test2` (`id` BIGINT, `dt` DATE, `state` STRING)
         > USING parquet
         > OPTIONS (
         >   `serialization.format` '1'
         > )
         > PARTITIONED BY (state);
Time taken: 0.06 seconds
spark-sql> insert into table skew_test2  select * from skew_test where state is not null;
Time taken: 1.208 seconds
spark-sql> insert into table skew_test2  select /*+ REPARTITION(100) */ * from skew_test where state is  null;
Time taken: 1.39 seconds

您应该看到 Spark 为最终语句创建了 100 个任务，并且您的 state=__HIVE_DEFAULT_PARTITION__ 目录应该包含 100 个 parquet 文件。有关 Spark-SQL 提示的更多信息，请查看https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-hint-framework.html#specifying-query-hints

【讨论】：

很遗憾，我正在使用目前无法更新的 SQL 代码库。所以这个解决方案可能有效，但我无法将它翻译成 SQL。不过，作为一个附带问题，由于在您分区时列似乎已删除（例如，如果您直接读取 state=FL 中的文件，它没有状态列。当您读取完整目录时，这似乎是从分区推断出来的），在直接写入默认分区之前是否需要删除 state 列？
我将编辑答案以展示如何通过 Spark-SQL 执行此操作。在我使用 Dataset API 的第一个示例中，您不一定需要删除 state，但您是正确的，因为这是一个好主意。在 SQL 中，您确实需要删除该列，正如我的示例所反映的那样。