使用Pandas UDF的PySpark序列计数

Question

我有一个列数据集，它对同一个月的任何列都增加值，然后在下个月重置。在

+----------+------+-----------+----+-----------+------------+
|      Date|column|column_2   |co_3|column_4   |column_5    |
+----------+------+-----------+----+-----------+------------+
|2016-12-14|     0|          0|   0|         14|           0|
|2016-12-14|     0|          0|   0|         14|           0|
|2016-12-14|     0|          0|   0|         18|           0|
|2016-12-14|     0|          0|   0|         19|           0|
|2016-12-14|     0|          0|   0|         20|           0|
|2016-12-14|     0|          0|   0|         26|           0|
|2016-12-14|     0|          0|   0|         60|           0|
|2016-12-14|     0|          0|   0|         63|           0|
|2016-12-14|     0|          0|   0|         78|           0|
|2016-12-14|     0|          0|   0|         90|           0|
+----------+------+-----------+----+-----------+------------+

问题是他们的日期总是一样的，所以我想做一些计数，然后在我们接近另一天时重新设置计数。在

我编写了一个Pandas UDF函数：

^{pr2}$

但是，输出似乎不是连续的：

sdf.filter(sdf.Date == "2016-12-14").sort("Date_Count").show()

+------------+----------+------+-----------+----+-----------+------------+---------+----------+--------+----------+-----+----------+
|Date_Convert|      Date|column|column_____|col_|column_____|column______|Date_Year|Date_Month|Date_Day|Date_Epoch|count|Date_Count|
+------------+----------+------+-----------+----+-----------+------------+---------+----------+--------+----------+-----+----------+
|  2016-12-14|2016-12-14|     0|          0|   0|         14|           0|     2016|        12|      14|1481673600|14504|         0|
|  2016-12-14|2016-12-14|     0|          0|   0|         18|           0|     2016|        12|      14|1481673600|14504|         0|
|  2016-12-14|2016-12-14|     0|          0|   0|         14|           0|     2016|        12|      14|1481673600|14504|         1|
|  2016-12-14|2016-12-14|     0|          0|   0|         18|           0|     2016|        12|      14|1481673600|14504|         1|
|  2016-12-14|2016-12-14|     0|          0|   0|         18|           0|     2016|        12|      14|1481673600|14504|         2|
|  2016-12-14|2016-12-14|     0|          0|   0|         14|           0|     2016|        12|      14|1481673600|14504|         2|
|  2016-12-14|2016-12-14|     0|          0|   0|         14|           0|     2016|        12|      14|1481673600|14504|         3|
+------------+----------+------+-----------+----+-----------+------------+---------+----------+--------+----------+-----+----------+

这是意料之中的，因为我猜数据帧被分成不同的机器（在DataBrick的社区版上有一些），并且每个机器都有自己的数组来维护。在

有没有顺序计数的方法？在

Answer 1

组合使用Window和row_number函数应该可以解决这个问题。正如你所说，我已经使用了所有列进行排序

dataset that has increasing values any columns for the same month...

您只能使用一个或多个值。在

from pyspark.sql import window as w
windowSpec = w.Window.partitionBy("Date").orderBy("column", "column_2", "co_3", "column_4", "column_5")

from pyspark.sql import functions as f
df.withColumn('inc_count', f.row_number().over(windowSpec)).show(truncate=False)

它应该给你

^{pr2}$