一种解决方案是使用repartition对 Dataframe 进行一次 Shuffle ，然后使用cache将结果保存在内存中：

cached_df_sales = df_sales.repartition("product_id").cache()
# and then do your work
cached_df_sales\
    .join(df_products,cached_df_sales.product_id==df_products.product_id,"inner")
cached_df_sales.groupby("product_id").agg(sum("num_pieces_sold"))

然而，我不确定这是不是一个好主意。缓存整个df_sales Dataframe 可能会占用大量内存，这取决于它的大小。而且，groupBy只会重排 Dataframe 的两列，这可能会相当便宜。在试图避免重排之前，我将首先确保这一点。
更一般地说，在尝试优化任何东西之前，简单地编写它，运行它，看看哪些东西需要时间，并专注于此。