如何在 sparklyr 中按 3 分钟时间戳聚合数据？

Question

我正在使用 sparklyr 进行一些快速分析。我在使用时间戳时确实遇到了一些问题。我有两个不同的数据帧：一个是 1 分钟间隔的行，另一个是 3 分钟间隔的行。

第一个数据集：（1 分钟间隔）

id  timefrom    timeto  value
10  "2017-06-06 10:30:00"   "2017-06-06 10:31:00"   50
10  "2017-06-06 10:31:00"   "2017-06-06 10:32:00"   80
10  "2017-06-06 10:32:00"   "2017-06-06 10:33:00"   20
22  "2017-06-06 10:33:00"   "2017-06-06 10:34:00"   30
22  "2017-06-06 10:34:00"   "2017-06-06 10:35:00"   50
22  "2017-06-06 10:35:00"   "2017-06-06 10:36:00"   50

第二个数据集：（3 分钟间隔）

id  timefrom    timeto  value
10  "2017-06-06 10:30:00"   "2017-06-06 10:33:00"   30
22  "2017-06-06 10:33:00"   "2017-06-06 10:36:00"   67
32  "2017-06-06 10:36:00"   "2017-06-06 10:39:00"   28
14  "2017-06-06 10:39:00"   "2017-06-06 10:42:00"   30
27  "2017-06-06 10:42:00"   "2017-06-06 10:55:00"   90

为了比较这 2 个数据集的值，我必须将第一个数据集聚合 3 分钟并计算值的平均值。此外，我必须从第二个数据集中找到最合适的window。

结果应如下所示：

id  timefrom    timeto  value1  value2
10  "2017-06-06 10:30:00"   "2017-06-06 10:33:00"   30  50
22  "2017-06-06 10:33:00"   "2017-06-06 10:36:00"   67  43.3

只用sparklyr可以实现吗？感谢您的帮助！

Answer 1

假设您的数据已经解析：

df1
# # Source:   table<df1> [?? x 4]
# # Database: spark_connection
#      id timefrom            timeto              value
#   <int> <dttm>              <dttm>              <int>
# 1    10 2017-06-06 08:30:00 2017-06-06 08:31:00    50
# 2    10 2017-06-06 08:31:00 2017-06-06 08:32:00    80
# 3    10 2017-06-06 08:32:00 2017-06-06 08:33:00    20
# 4    22 2017-06-06 08:33:00 2017-06-06 08:34:00    30
# 5    22 2017-06-06 08:34:00 2017-06-06 08:35:00    50
# 6    22 2017-06-06 08:35:00 2017-06-06 08:36:00    50

df2
# # Source:   table<df2> [?? x 4]
# # Database: spark_connection
#      id timefrom            timeto              value
#   <int> <dttm>              <dttm>              <int>
# 1    10 2017-06-06 08:30:00 2017-06-06 08:33:00    30
# 2    22 2017-06-06 08:33:00 2017-06-06 08:36:00    67
# 3    32 2017-06-06 08:36:00 2017-06-06 08:39:00    28
# 4    14 2017-06-06 08:39:00 2017-06-06 08:42:00    30
# 5    27 2017-06-06 08:42:00 2017-06-06 08:55:00    90

你可以使用 window function:

exprs <- list(
  "id", "value as value2",
  # window generates structure struct<start: timestamp, end: timestamp>
  # we use dot syntax to access nested fields
  "window.start as timefrom", "window.end as timeto")

df1_agg <- df1 %>% 
  mutate(window = window(timefrom, "3 minutes")) %>% 
  group_by(id, window) %>% 
  summarise(value = avg(value)) %>%
  # As far as I am aware there is no sparklyr syntax 
  # for accessing struct fields, so we'll use simple SQL expression
  spark_dataframe() %>% 
  invoke("selectExpr", exprs) %>% 
  sdf_register() %>%
  print()

# Source:   table<sparklyr_tmp_472ee8ba244> [?? x 4]
# Database: spark_connection
     id value2 timefrom            timeto             
  <int>  <dbl> <dttm>              <dttm>             
1    22   43.3 2017-06-06 08:33:00 2017-06-06 08:36:00
2    10   50.0 2017-06-06 08:30:00 2017-06-06 08:33:00

然后你可以通过 id 和时间戳列：

df2 %>% inner_join(df1_agg, by = c("id", "timefrom", "timeto"))
# # Source:   lazy query [?? x 5]
# # Database: spark_connection
#      id timefrom            timeto              value value2
#   <int> <dttm>              <dttm>              <int>  <dbl>
# 1    10 2017-06-06 08:30:00 2017-06-06 08:33:00    30   50.0
# 2    22 2017-06-06 08:33:00 2017-06-06 08:36:00    67   43.3

如何在 sparklyr 中按 3 分钟时间戳聚合数据？

How to aggregate Data by 3 minutes timestamps in sparklyr?

r

apache-spark

dplyr

sparklyr