如何使用 Spark 数据框获取每周和每小时发生的行数？

Question

我想返回一周中每个小时发生事件的次数（按天和小时分组）。使用 Spark DataFrame，我可以获得带有“dateOccurred”字符串列的返回行的列表。 （24号和17号是周五，23号和16号是周四）

• 行（日期发生='24-04-2020 10:08:00'）

• 行（日期发生='24-04-2020 11:52:00'）

• 行（日期发生='24-04-2020 11:35:00'）

• 行（日期发生='23-04-2020 15:13:00'）

• 行（日期发生='23-04-2020 15:20:00'）

• 行（日期发生='23-04-2020 23:52:00'）

• 行（日期发生='16-04-2020 15:22:00'）

• 行（日期发生='16-04-2020 23:12:00'）

• 行（日期发生='16-04-2020 14:28:00'）

• 行（日期发生='17-04-2020 10:16:00'）

• 行（日期发生='17-04-2020 11:19:00'）

• 行（日期发生='17-04-2020 12:52:00'）

我想将结果转换为以下格式”

• ('星期五_10', 2)
• ('星期五_11', 3)
• ('星期五_12', 1)
• ('星期四_15', 3)
• ('星期四_23', 2)
• ('星期四_14', 1)

Answer 1

您必须从日期列中提取日期和小时。今天，您有两种选择。首先，您可以使用 UDF 提取工作日全名。如果天作为整数就足够了，您可以使用内置的 PySpark 功能。之后，您可以连接两列并执行 groupBy + count

import datetime
import pyspark.sql.functions as f
from pyspark.sql.types import StringType

def  get_day_from_date(dt)
    dt_parsed = datetime.datetime.strptime(dt_2, '%d-%m-%Y %H:%M:%S')
    ans = datetime.date(dt_parsed.year, dt_parsed.month, dt_parsed.day)
    return ans.strftime("%A")

to_day = f.udf(get_day_from_date, StringType())
df = df.withColumn('dateOccurred_ts', f.to_timestamp('Timestamp', 'dd-MM-yyyy HH:mm:ss'))

# udf returns Weekday as locale’s full name
df = df.withColumn('day', to_day(f.col('dateOccurred')))
## other solution only returns day of the week of a given date as integer.
df = df.withColumn('day', f.dayofweek('dateOccurred_ts'))

df = df.withColumn('hour', f.hour('dateOccured_ts'))
df_2 = df.select(f.concat_ws('_', f.col('day'), f.col('hour')))
df = df.groupBy('day_hour').agg(f.count(f.lit(1)).alias('dateOccurred_cnt'))