Spark - 按时间戳范围排序的统计计算

Question

我正在尝试根据一系列小时和\或天数计算统计量度。

意思是，我有一个类似这样的 CSV 文件：

TRANSACTION_URL    START_TIME        END_TIME           SIZE    FLAG
www.google.com     20170113093210    20170113093210     150      1
www.cnet.com       20170113114510    20170113093210     150      2

START_TIME 和 END_TIME 采用 yyyyMMddhhmmss 格式。

我首先使用以下代码将其转换为yyyy-MM-dd hh:mm:ss 格式：

from_pattern = 'yyyyMMddhhmmss'
to_pattern = 'yyyy-MM-dd hh:mm:ss'

log_df = log_df.withColumn('START_TIME', from_unixtime(unix_timestamp(
    log_df['START_TIME'].cast(StringType()), from_pattern), to_pattern).cast(TimestampType()))

然后，我想使用groupBy() 来计算，例如，基于事务时间框架的 SIZE 列的平均值。

例如，我想做这样的事情：

for all transactions that are between 09:00 to 11:00
    calculate SIZE mean

for all transactions that are between 14:00 to 16:00
    calculate SIZE mean

还有：

for all transactions that are in a WEEKEND date
    calculate SIZE mean

for all transactions that are NOT in a WEEKEND date
    calculate SIZE mean

我知道如何将 groupBy 用于“默认”配置，例如根据 FLAG 列值计算 SIZE 列的统计度量。我正在使用类似的东西：

log_df.cache().groupBy('FLAG').agg(mean('SIZE').alias("Mean"), stddev('SIZE').alias("Stddev")).\
    withColumn("Variance", pow(col("Stddev"), 2)).show(3, False)

所以，我的问题是：

1. 如何在一个小时范围内实现这样的分组和计算？ （第一个伪代码示例）

2. 如何按日期实现这样的分组和计算？ （第二个伪代码示例）

如果是周末约会，有没有可以接收yy-MM-dd并返回true的python包？

谢谢

Answer 1

假设您有一个函数 encode_dates 接收日期并返回您感兴趣的所有时间段的编码序列。例如，对于 9-12 星期二，它将返回 Seq("9-11","10 -12","11-13","工作日")。这将是一个常规的 scala 函数（与 spark 无关）。

现在您可以将其设为 UDF 并将其添加为列并展开列，这样您将拥有多个副本。现在您需要做的就是为 groupby 添加此列。

所以它看起来像这样：

val encodeUDF = udf(encode_dates _)
log_df.cache().withColumn("timePeriod", explode(encodeUDF($"start_date", $"end_date").groupBy('FLAG', 'timePeriod').agg(mean('SIZE').alias("Mean"), stddev('SIZE').alias("Stddev")).
withColumn("Variance", pow(col("Stddev"), 2)).show(3, False)