Spark Dataframes 中的分区和集群答案

【问题标题】：Partition and cluster by in Spark DataframesSpark Dataframes 中的分区和集群
【发布时间】：2018-02-01 09:47:04
【问题描述】：

我们是否可以在 Spark 中按列进行分区，然后按另一列进行聚类？

在我的示例中，我在包含数百万行的表中有一个 month 列和一个 cust_id 列。我可以说当我将 DataFrame 保存到 hive 表以基于月份将表分区并按cust_id 聚类为 50 个文件时？

忽略cust_id的聚类，这里有三种不同的选择

df.write.partitionBy("month").saveAsTable("tbl")
df.repartition(100).write.partitionBy("month").saveAsTable("tbl")
df.repartition("month").write.saveAsTable("tbl")

第一种情况和最后一种情况在 Spark 中的作用相似，但我假设它只是在 Hive 中以不同的方式写入数据（文件夹而不是每个月的文件）。

在第二个选项中，重新分区由partitionBy 撤消不是吗？

至少我怎样才能避免这种情况？

是否甚至可以通过在 Spark 中指定多个存储桶来通过高基数列进行重新分区？

【问题讨论】：

标签： apache-spark apache-spark-sql

【解决方案1】：

我们是否可以在 Spark 中按列进行分区，然后按另一列进行聚类？

有可能，但repartition 在这里帮不了你。

df.write
  .partitionBy("month")
  .clusterBy(50, "id")
  .saveAsTable("tbl")

等价的：

CREATE TABLE users_bucketed_and_partitioned(
 month T,
 id U
) USING parquet 
PARTITIONED BY (month)
CLUSTERED BY(id) INTO 50 BUCKETS;

请记住，它与 Hive 不兼容，并且似乎具有如此意想不到的性能影响。

【讨论】：

这适用于哪个版本的 spark？我正在使用 spark 2.1 并在 pyspark 上找不到它。这仅在scala上吗？错误：“DataFrameWriter”对象没有属性“clusterBy”
在 Python 2.3 (issues.apache.org/jira/browse/SPARK-16931) 中，但它很容易打补丁（检查 PR 以了解实现），如果你不想打补丁，你总是可以使用 SQL (spark.sql(...)) .
我可以看到 bucketBy API 自 2.1 以来在 scala 中可用并且可以使用它，但我很好奇如何在 spark.sql 中完成此操作？您的意思是将创建和插入分成两步并像在 SQL 中一样手动完成吗？
df.createOrReplaceTempView("a_view") 然后CREATE TABLE ... AS SELECT * FROM a_view
只是评论，spark上的cluster by method有点乱。它为大型流创建了数千个文件，因为每个执行程序会生成 n 个文件（每个存储桶一个），因此您最终可能会得到 n*exec_count 个文件。

【解决方案2】：

只是为了让其他不想修补或编写 SQL 插入语句的人知道，但是在数据帧上使用 repartition 和 partitionBy 实际上可以按我的意愿工作，而不是按我的预期工作。

意思是先按key分区，再按number重新分区。

例子：

df.repartition(100).write.partitionBy("month").saveAsTable("tbl")

在每个分区内生成 100 个大小大致相等的文件，其中为 hive 上的结果表中的每个不同月份值创建一个文件夹（分区）。

【讨论】：