为了扩展，您需要改为按id聚合，而不收集任何结果。请尝试以下方法：

import pyspark.sql.functions as F
from pyspark.sql import Window

data = spark.createDataFrame([(1, "ID_1"), (1, "ID_2"), (1, "ID_3"),
                              (2, "ID_1"), (2, "ID_4"), (2, "ID_5"),
                              (3, "ID_6")], ["Week", "ID"])

win = Window.partitionBy('ID').orderBy('Week')

agg_data = (
  data
  .withColumn("prevWeek", F.lag("Week", offset=1).over(win))
  .withColumn("isInPrevWeek", 
              F.col("prevWeek").isNotNull() & ((F.col("Week") - F.col("prevWeek")) == 1))
  .filter(~F.col("isInPrevWeek"))
  .groupBy("Week")
  .agg(F.count("*").alias("newIDs"),
       F.array_sort(F.collect_list("ID")).alias("showNewIDs"))  # Remove in production
  .orderBy("Week")
)

agg_data.show()

第一，功能 lag 使用一个窗口函数创建前一周的新列，该函数允许单独考虑每个id（分区）并按时间顺序对周进行排序。这可以很好地扩展，因为spark任务是由一组id组成的。
那么， isInPrevWeek 检查身份证是否确实在前一周。如果是这样，则过滤掉记录。现在，你只需要按周计算重新录制的ID。

+----+------+------------------+
|Week|newIDs|        showNewIDs|
+----+------+------------------+
|   1|     3|[ID_1, ID_2, ID_3]|
|   2|     2|      [ID_4, ID_5]|
|   3|     1|            [ID_6]|
+----+------+------------------+

请注意，代码段收集id只是为了说明目的，但这不是计算所必需的。