Spark高效的groupby操作-重新分区？

Question

我正在使用 pyspark 2.3，我正在尝试找出从数据框中获取一些汇总统计信息的最有效方法。

我有一个包含 15 亿条记录的数据框，分布在一个由 10 个节点组成的相对较小的集群中。每个都有 16gb 的 ram 和 4 个内核。我的复制因子设置为 2。

我的数据框可能有 15 列，它们是数据类型的混合，但我只对两列感兴趣 - ID 和 eventDate。我想运行的代码很简单：

output = df.groupby(['ID']).agg(F.min('eventDate').alias("firstDate"),F.max('eventDate').alias("lastDate"))
output.write.parquet('hdfs:///somewhere/dateFile.parquet',mode='overwrite')

我想弄清楚的是执行此操作的最有效方法。 ID，即我分组依据的字段，有 12m 个值，df.rdd.getNumPartitions() 目前是 642。

我最好先将我的数据框投影到我想要的两列吗？有这么多 ID，我应该先重新分区我的数据集吗？我应该删除重复项吗？我可以在我的 groupby 之前运行这样的东西：

df = df[['ID','eventDate']].drop_duplicates().repartition(x)

或

df = df[['ID','eventDate']].repartition(x)

我正在努力弄清楚什么会优化运行时。任何有关预先确定运行时的指导将不胜感激。如果可能的话，我不希望只是“测试一下”，因为我有几个这样的查询要运行，每个都需要一段时间。

Answer 1

这可能不是您正在寻找的答案，但此操作的最佳代码正是

output = df.groupby(['ID']). \
 agg(F.min('eventDate').alias("firstDate"), F.max('eventDate').alias("lastDate"))
output.write.parquet('hdfs:///somewhere/dateFile.parquet', mode='overwrite')

Spark 通过仅首先选择整个操作所需的必要列来优化流程。然后 Spark 按ID 对您的数据进行分区，并在每个分区上开始聚合过程。

允许最大数量的执行者肯定会有所帮助。我建议（根据您的描述）设置spark.executor.instances=10; spark.executor.memory=10g。 12m 个值是一个合理的数量，也许尝试增加 shuffle 分区的数量，f.e. spark.sql.shuffle.partitions=400，这样您就不会遇到一些烦人的内存开销异常。

Answer 2

@flyingmeatball，

在进行聚合之前，请执行以下步骤

1 - 删除不需要的数据（它会吃掉你的资源）。

2-根据您的数据重新分区和缓存数据（它将消除执行时间）

提示：如果数据来自 Cassandra，则按分区键重新分区数据，以避免数据混洗

现在你可以使用聚合逻辑了；）

谢谢，
维马利什