在这种情况下，我会使用SparkSQL API，因为windows将非常有用。对于每一年，让我们计算实现至少65%销售额所需的产品百分比：

# let's create some sample data. I assume we have one line per sale
df = spark.createDataFrame([('2020', 'prod1'), ('2020', 'prod1'), ('2020', 'prod1'),
    ('2020', 'prod1'), ('2020', 'prod1'), ('2020', 'prod3'), ('2020', 'prod1'),
    ('2020', 'prod2'),  ('2020', 'prod2'),  ('2020', 'prod3'),  ('2020', 'prod4'),
    ('2020', 'prod5')], ['year', 'asin'])

# let's start by counting the number of sales per product, per year
df.groupBy("year", "asin").count().show()

+----+-----+-----+
|year| asin|count|
+----+-----+-----+
|2020|prod1|    6|
|2020|prod3|    2|
|2020|prod2|    2|
|2020|prod4|    1|
|2020|prod5|    1|
+----+-----+-----+

现在，让我们使用windows来计算回答您的问题所需的内容：

• 每年产品总数：product_count
• 每年总销售量：total_sales
• 累计销售数量，从销售最多的产品开始：cum_sales
• 产品指数，从销售最多的产品开始：product_index From there, product_per is the percentage of product, sales_per the percentage of sales so that we can see if at least 65% of the sales were made by less than 20% of the products. We can finally compute dist'，销售百分比与65%之间的距离。我们将使用该列保留允许达到65%以上销售额的第一行。

一个二个一个一个
因此，在本例中，我们需要40%的产品（5个中的2个）才能达到至少65%的销售额，我们只保留这一行：

dist_win = Window.partitionBy("year").orderBy("dist")
rich_df.where(f.col("dist") >= 0)\
    .withColumn("dist_rank", f.rank().over(dist_win))\
    .where(f.col("dist_rank") == 1)\
    .select("year", "product_per", "sales_per", (f.col("product_per") < 0.2).alias("hasLongTail"))\
    .show()

+----+-----------+------------------+-----------+
|year|product_per|         sales_per|hasLongTail|
+----+-----------+------------------+-----------+
|2020|        0.4|0.6666666666666666|      false|
+----+-----------+------------------+-----------+

这将持续一年以上-）