GroupBy 在 PySpark DataFrame 问题上使用时间频率答案

【问题标题】：GroupBy using Time Frequency on PySpark DataFrame IssueGroupBy 在 PySpark DataFrame 问题上使用时间频率
【发布时间】：2020-11-20 03:43:31
【问题描述】：

我是 PySpark 的新手。

我正在尝试执行 GroupBy 操作以获取聚合计数。但我无法根据时间频率执行 groupBy 。我需要使用“CAPTUREDTIME、NODE、CHANNEL、LOCATION、TACK”字段执行“groupBy”。但是在这个 groupBy 中，我应该使用“CAPTUREDTIME”字段根据“hourly”、“daily”、“weekly”、“monthly”进行分组。

请找到以下示例数据。

-----------------+------+------+--------+----------+--------------

|CAPTUREDTIME|      NODE|       CHANNEL  |  LOCATION|    TACK

+-----------------+------+------+--------+----------+-------------

|20-05-09 03:06:21|   PUSC_RES|   SIMPLEX|  NORTH_AL|    UE220034

|20-05-09 04:33:04|   PUSC_RES|   SIMPLEX|  SOUTH_AL|    UE220034

|20-05-09 12:04:52|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057

|20-05-10 04:24:09|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057

|20-05-10 04:33:04|   PUSC_RES|   SIMPLEX|  SOUTH_AL|    UE220034

|20-04-09 10:57:48|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057

|20-04-09 12:12:26|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057

|20-04-09 03:26:33|   PUSC_RES|   SIMPLEX|  NORTH_AL|    UE220071

+-----------------+------+------+--------+----------+-------------

我使用了下面的 pyspark 代码

df = df.groupby("CAPTUREDTIME", "NODE", "CHANNEL", "LOCATION", "TACK").agg(
    func.count("TACK").alias("count")
)

如何将上述代码扩展到“每小时”、“每天”、“每周”、“每月”分组？

我需要以下格式的输出（共享示例输出）：

每小时：

|捕获时间|节点|频道 |位置|大头钉|计数

|20-05-09 03:00:00| PUSC_RES|简单| NORTH_AL| UE220034| 2

|20-05-09 04:00:00| PUSC_RES|简单|南_阿尔| UE220034| 2

每天：

|捕获时间|节点|频道 |位置|大头钉|计数

|20-05-09 00:00:00| PUSC_RES|简单| NORTH_AL| UE220034| 1

|20-05-09 00:00:00| PUSC_RES|简单|南_阿尔| UE220034| 2

|20-05-09 00:00:00| TESC_RES|简单| NORTH_AL| UE220057| 3

每周：

|捕获时间|节点|频道 |位置|大头钉|计数

|20-05-09 00:00:00| PUSC_RES|简单| NORTH_AL| UE220034| 1

每月：

|捕获时间|节点|频道 |位置|大头钉|计数

|20-05-09 00:00:00| PUSC_RES|简单| NORTH_AL| UE220034| 1

【问题讨论】：

看看stackoverflow.com/questions/30949202/…
您需要从您的日期时间列中创建您想要的列，即“hourly”、“daily”、“weekly”、“monthly”，然后分组
@LarsSkaug：该链接说明了如何拆分字段。在这里，我需要对现有字段执行 groupby。它没有解决这个问题
预期的答案是什么？能否请您以表格格式分享以便更好地理解

标签： pyspark apache-spark-sql pyspark-dataframes

【解决方案1】：

您有两种方法来回答您的问题，要么将时间戳转换为要分组的日期粒度，要么（如您在 cmets 中所说）使用 sql 窗口函数按您想要的间隔分组.

只知道通过 Spark 中的窗口 SQL 函数无法实现每月聚合。

这里你可以看到代码，前三个例子使用了window SQL函数，最后一个例子每月转换时间戳，然后按每列分组。

df = spark.createDataFrame(
    [
        ("20-05-09 03:06:21", "PUSC_RES", "SIMPLEX", "NORTH_AL", "UE220034"),
        ("20-05-09 04:33:04", "PUSC_RES", "SIMPLEX", "SOUTH_AL", "UE220034"),
        ("20-05-09 12:04:52", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-05-10 04:24:09", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-05-10 04:33:04", "PUSC_RES", "SIMPLEX", "SOUTH_AL", "UE220034"),
        ("20-04-09 10:57:48", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-04-09 12:12:26", "TESC_RES", "SIMPLEX", "NORTH_AL", "UE220057"),
        ("20-04-09 03:26:33", "PUSC_RES", "SIMPLEX", "NORTH_AL", "UE220071")
    ],
    ['CAPTUREDTIME', 'NODE', 'CHANNEL', 'LOCATION', 'TACK']
)

from pyspark.sql.functions import col, count, date_format, date_sub, date_trunc, month, next_day, to_timestamp, weekofyear, window, year

每小时

我仍然只为这个保留窗口逻辑，因此我们可以为每个人参考 Spark 中的每一种可能性。在显示数据框之前，我只在末尾选择窗口的开头。

hourly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .groupBy(window(col("captured_time"), "1 hour").alias("captured_time"), "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
    .withColumn("captured_time_hour", col("captured_time.start"))
    .drop("captured_time")
)
hourly.sort("captured_time_hour").show(100, False)

每天

通过date_trunc 函数，我可以仅考虑日期截断时间戳

daily = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("captured_time_day", date_trunc("day", col("captured_time")))
    .groupBy("captured_time_day", "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
)
daily.sort("captured_time_day").show(100, False)

每周

这个有点棘手。首先，我在星期一使用 next_day 函数。如果您认为星期日是一周的开始，请根据它更新此代码，但我认为星期一是一周的开始（这取决于我相信的 SQL 方言和地区）

那么我们也可以添加一个weekofyear函数来检索你想要的周数

weekly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("start_day", date_sub(next_day(col("captured_time"), "monday"), 7))
    .groupBy("start_day", "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*"))
    .withColumn("start_day", to_timestamp(col("start_day")))
    .withColumn("week_of_year", weekofyear(col("start_day")))
)
weekly.sort("start_day").show(100, False)

每月

我们只是将时间戳格式化为日期，然后将其转换回时间戳。这样做只是为了展示另一种方法。我们可以将时间戳截断为日常用例。我还展示了两种提取月份名称和缩写的方法。请注意您的 Spark 版本，因为这是在 Spark 3.0.0 中测试的

monthly = (
    df
    .withColumn("captured_time", to_timestamp(col('CAPTUREDTIME'), 'yy-MM-dd HH:mm:ss'))
    .withColumn("captured_time_month", date_format(col('captured_time'), '1/M/yyyy'))
    .groupBy(col("captured_time_month"), "NODE", "CHANNEL", "LOCATION", "TACK")
    .agg(count("*").alias("Count TACK"))
    .withColumn("captured_time_month", to_timestamp(col("captured_time_month"), '1/M/yyyy'))
    .withColumn("month", month(col("captured_time_month")))
    .withColumn("month_abbr", date_format(col("captured_time_month"),'MMM'))
    .withColumn("full_month_name", date_format(col("captured_time_month"),'MMMM'))
)
monthly.sort("captured_time_month").show(100, False)

咻！

【讨论】：

你好@tricky：我怎样才能得到每周 groupby 的周数？此外，对于每日、每周、每月聚合，“HH:mm:ss”应为“00:00:00”。
嘿@stacktesting 我刚刚编辑了代码并根据您想要的内容发布。干杯：D
谢谢@tricky。有效。但是当我将每周的日期从“星期一”更改为“星期日”时。它不工作。我在这里错过了什么？

【解决方案2】：

Spark 为日期操作提供了一个相对丰富的库。您的问题的答案是提取日期部分和显示日期格式的组合。

我重新创建了您的数据，如下所示：

val capturesRaw = spark.read
  .option("ignoreLeadingWhiteSpace", "true")
  .option("ignoreTrailingWhiteSpace", "true")
  .option("delimiter", "|")
  .option("header", "true")
  .csv(spark.sparkContext.parallelize("""
      CAPTUREDTIME|      NODE|       CHANNEL  |  LOCATION|    TACK
      20-05-09 03:06:21|   PUSC_RES|   SIMPLEX|  NORTH_AL|    UE220034
      20-05-09 04:33:04|   PUSC_RES|   SIMPLEX|  SOUTH_AL|    UE220034
      20-05-09 12:04:52|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057
      20-05-10 04:24:09|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057
      20-05-10 04:33:04|   PUSC_RES|   SIMPLEX|  SOUTH_AL|    UE220034
      20-04-09 10:57:48|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057
      20-04-09 12:12:26|   TESC_RES|   SIMPLEX|  NORTH_AL|    UE220057
      20-04-09 03:26:33|   PUSC_RES|   SIMPLEX|  NORTH_AL|    UE220071"""
  .split("\n")).toDS)

注意：我使用的是 Scala，但代码差异很小，希望您能理解。我相信开头的val实际上是唯一的区别。

我假设前两位数字代表两位数的年份？要继续，我们需要确保捕获时间是一个时间戳。我更喜欢使用 SQL 来操作数据帧，因为我发现它更具可读性。

spark.sql("""select to_timestamp('20' || capturedtime) capturedtime, NODE, CHANNEL, 
             LOCATION, TACK from captures_raw""")
    .createOrReplaceTempView("captures_raw")

如果您愿意，可以直接在数据帧上完成相同的操作

capturesRaw.withColumn("capturedtimestamp",
        to_timestamp(col("capturedtime"), "yy-MM-dd hh:mm:ss"))

此时，我们可以创建您请求的字段：

spark.sql("""select capturedtime,
           month(capturedtime) cap_month, 
           weekofyear(capturedtime) cap_week,   
           day(capturedtime) cap_day, 
           hour(capturedtime) cap_hr, NODE, CHANNEL, LOCATION, TACK 
           from captures_raw""").createOrReplaceTempView("captures")

创建完字段后，我们就可以回答您的问题了。例如，要仅按月份聚合（没有其余时间戳），请执行以下操作：

spark.sql("""select date_format(capturedtime, "yyyy-MM") year_month, cap_month,
             cap_week, cap_day, cap_hr, count(*) count
             from captures 
             group by 1,2,3,4,5""").show

+----------+---------+--------+-------+------+-----+
|year_month|cap_month|cap_week|cap_day|cap_hr|count|
+----------+---------+--------+-------+------+-----+
|   2020-04|        4|      15|      9|     3|    1|
|   2020-04|        4|      15|      9|    10|    1|
|   2020-05|        5|      19|      9|     4|    1|
|   2020-05|        5|      19|      9|    12|    1|
|   2020-04|        4|      15|      9|    12|    1|
|   2020-05|        5|      19|      9|     3|    1|
|   2020-05|        5|      19|     10|     4|    2|
+----------+---------+--------+-------+------+-----+

可以按如下方式生成每日摘要：

spark.sql("""select date_format(capturedtime, "yyyy-MM-dd") captured_date,
             cap_day, cap_hr, count(*) count
             from captures 
             group by 1,2,3""").show

+-------------+-------+------+-----+
|captured_date|cap_day|cap_hr|count|
+-------------+-------+------+-----+
|   2020-05-10|     10|     4|    2|
|   2020-04-09|      9|    12|    1|
|   2020-05-09|      9|     4|    1|
|   2020-05-09|      9|    12|    1|
|   2020-04-09|      9|     3|    1|
|   2020-04-09|      9|    10|    1|
|   2020-05-09|      9|     3|    1|
+-------------+-------+------+-----+

【讨论】：

在上面编写的查询中，我们对各个时间频率执行 groupBy，这很好。但是如何在每行的输出中获取“CAPTURED TIME”的原始值呢？
意味着在每日聚合中，“捕获时间”的值应该是单独的日期，对于每周，应该是特定的一周，
我用每月和每日总结的例子更新了我的答案。
我们可以不使用窗口功能吗？问题是什么 df = df.groupby(window('CAPTUREDTIME','60 minutes'), "NODE", "CHANNEL", "LOCATION", "TACK").agg( func.count("TACK") .alias("count") df = df.groupby(window('CAPTUREDTIME','1 day'), "NODE", "CHANNEL", "LOCATION", "TACK").agg( func.count("TACK ").alias("count") df = df.groupby(window('CAPTUREDTIME','1 week'), "NODE", "CHANNEL", "LOCATION", "TACK").agg( func.count( "TACK").alias("count") df = df.groupby(window('CAPTUREDTIME','1 month'), "NODE", "CHANNEL", "LOCATION", "TACK").agg( func. count("TACK").alias("count")
没有。正在检查窗口选项。但是不知道这种情况下窗口函数的作用