PySpark 非重复列数

Question

我有一个如下所示的 PySpark DataFrame：

+------+-----------+
|src_ip|  timestamp|
+------+-----------+
|A     |2020-06-19 |
|B     |2020-06-19 |
|B     |2020-06-20 |
|C     |2020-06-20 |
|D     |2020-06-21 |
+------+-----------+

我想检索每个不同 IP 地址的计数，这些 IP 地址被细分为每天看到多少个不同 IP 地址。

我试过：

df.groupBy(window(df['timestamp'], "1 day")) \
           .agg(countDistinct('src_ip')) \
           .orderBy("window").show()

但是，这并没有给我正确的结果，因为它将 DF 分成时间 windows，并获得每个时间的不同计数 windows，如图所示：

+-----------+-----------------------+
|  window   | count(DISTINCT(src_ip)|
+-----------+-----------------------+
|2020-06-19 | 2                     |
|2020-06-20 | 2                     |
|2020-06-21 | 1                     |
+-----------+-----------------------+

这是不正确的，因为 B 已经出现在 2020-06-19 上，应该被归类为不同的。

结果 table 我希望看到的是：

+-----------+-----------------------+
|  window   | count(DISTINCT(src_ip)|
+-----------+-----------------------+
|2020-06-19 | 2                     |
|2020-06-20 | 1                     |
|2020-06-21 | 1                     |
+-----------+-----------------------+

PySpark 甚至可以做到这一点吗？非常感谢任何帮助。

Answer 1

这是你想要的吗？或者请添加更多解释。

df.show(10, False)

+------+----------+
|src_ip|timestamp |
+------+----------+
|A     |2020-06-19|
|B     |2020-06-19|
|B     |2020-06-20|
|C     |2020-06-20|
|D     |2020-06-21|
+------+----------+


from pyspark.sql.functions import min, window, count

df.groupBy('src_ip').agg(min('timestamp').alias('timestamp')) \
  .groupBy('timestamp').agg(count('src_ip').alias('count')) \
  .orderBy('timestamp').show(10, False)

+----------+-----+
|timestamp |count|
+----------+-----+
|2020-06-19|2    |
|2020-06-20|1    |
|2020-06-21|1    |
+----------+-----+

PySpark 非重复列数

PySpark Distinct Count of Column

python

window

dataframe

apache-spark

pyspark