根据多个连续时间流计算行数答案

【问题标题】：Count rows based on multiple consecutive time flows根据多个连续时间流计算行数
【发布时间】：2016-07-06 19:01:00
【问题描述】：

我有一个大的时间序列数据文件，如下所示。数据集涵盖多年，增量为 15 分钟。一个小子集如下所示：

uniqueid     time
a            2014-04-30 23:30:00 
a            2014-04-30 23:45:00
a            2014-05-01 00:00:00
a            2014-05-01 00:15:00
a            2014-05-12 13:45:00
a            2014-05-12 14:00:00
b            2014-05-12 13:45:00
b            2014-05-12 14:00:00
b            2014-05-12 14:30:00

复制上面的内容：

time<-c("2014-04-30 23:30:00","2014-04-30 23:45:00","2014-05-01 00:00:00","2014-05-01 00:15:00",
    "2014-05-12 13:45:00","2014-05-12 14:00:00","2014-05-12 13:45:00","2014-05-12 14:00:00",
    "2014-05-12 14:30:00")

uniqueid<-c("a","a","a","a","a","a","b","b","b")
mydf<-data.frame(uniqueid,time)

我的目标是计算每个连续时间流中每个唯一 ID 的行数。连续时间跨度是指连续每 15 分钟标记一个唯一 ID（例如 id A，从 30.04.14 23.30 小时到 01.05.14 00.15 小时 - 因此为 4 行），但当此流为 15 -分钟迭代被中断（在 01.05.14 00:15 之后，它没有在 01.05.14 00:30 标记，因此它被中断），它应该将下一个时间戳记为新的连续时间流的开始，并再次计算行，直到此流程再次中断。时间是 POSIX。

如您在上面的示例中所见；连续的时间流可能涵盖不同的日子、不同的月份或不同的年份。我有许多唯一的 id（如前所述，一个非常大的文件），所以我正在寻找一种我的计算机可以处理的方式（循环可能不起作用）。

我正在寻找类似的输出：

uniqueid    flow     number_rows
a           1        4
a           2        2
b           3        2
b           4        1

我研究了一些时间包（例如 lubridate），但鉴于我有限的 R 知识，我什至不知道从哪里开始。

我希望一切都清楚 - 如果没有，我很乐意尝试进一步澄清。非常感谢您！

【问题讨论】：

标签： r time

【解决方案1】：

使用data.table 也使用时差的另一种方法是使用data.table 内部值作为组号和每个组中的行数：

library(data.table)
res<-setDT(mydf)[, list(number_rows=.N,flow=.GRP),
                 by=.(uniqueid,cumsum(as.numeric(difftime(time,shift(time,1L,type="lag",fill=0))) - 15))][,cumsum:=NULL]
print(res)

   uniqueid number_rows flow
1:        a           4    1
2:        a           2    2
3:        b           2    3
4:        b           1    4

此外，由于您发布的示例数据与您发布的子集不一致，我将我的数据包含在下面：

数据

time<-as.POSIXct(c("2014-04-30 23:30:00","2014-04-30 23:45:00","2014-05-01 00:00:00","2014-05-01 00:15:00",
        "2014-05-12 13:45:00","2014-05-12 14:00:00","2014-05-12 13:45:00","2014-05-12 14:00:00",
        "2014-05-12 14:30:00"))


uniqueid<-c("a","a","a","a","a","a","b","b","b")
mydf<-data.frame(uniqueid,time)

【讨论】：

不错！如果你 unclass(mydf$time) 并检查差异是 15*60 = 900L，它会快一点。
感谢您提供此解决方案！完美而快速地工作。并对错误的样本数据表示歉意，现已在帖子中进行了调整。
感谢 Arun 对速度的评论，将包括它以使其运行得更快。

【解决方案2】：

您可以按uniqueid 和不等于15 min 的行之间的时间差的累积总和进行分组，这给出了流id，然后行数应该可以满足您的需求：

逻辑的一个理由是，只要每个uniqueid内的时间差不等于15，就应该生成一个新的流程，因此我们将其标记为TRUE并将其与cumsum结合，它成为一个新的flow id，具有以下连续行：

library(dplyr)
mydf$time <- as.POSIXct(mydf$time, "%Y-%m-%d %H:%M:%S")
# convert the time column to POSIXct class so that we can apply the diff function correctly
mydf %>% group_by(uniqueid, flow = 1 + cumsum(c(F, diff(time) != 15))) %>% 
         summarize(num_rows = n())

# Source: local data frame [4 x 3]
# Groups: uniqueid [?]
# 
#   uniqueid  flow num_rows
#     <fctr> <dbl>    <int>
# 1        a     1        4
# 2        a     2        2
# 3        b     3        2
# 4        b     4        1

【讨论】：

感谢您的解决方案！然而，正如 ARobertson 所指出的，dplyr 对于这么大的数据集来说太慢了。此外，该解决方案运行良好。

【解决方案3】：

Base R 相当快。使用粗略的基准测试，我发现它在 DT 的一半时间内完成，我厌倦了等待 dplyr。

# estimated size of data, years x days x hours x 15mins x uniqueids
5*365*24*4*1000 # = approx 180M

# make data with posixct and characters of 180M rows, mydf is approx 2.5GB in memory
time<-rep(as.POSIXct(c("2014-04-30 23:30:00","2014-04-30 23:45:00","2014-05-01 00:00:00","2014-05-01 00:15:00",
        "2014-05-12 13:45:00","2014-05-12 14:00:00","2014-05-12 13:45:00","2014-05-12 14:00:00",
        "2014-05-12 14:30:00")),times = 20000000)

uniqueid<-rep(as.character(c("a","a","a","a","a","a","b","b","b")),times = 20000000)

mydf<-data.frame(uniqueid,time = time)
rm(time,uniqueid);gc()

基础R：

# assumes that uniqueid's are in groups and in order, and there won't be a followed by b that have the 15 minute "flow"
starttime <- Sys.time()

# find failed flows
mydf$diff <- c(0,diff(mydf$time))
mydf$flowstop <- mydf$diff != 15

# give each flow an id
mydf$flowid <- cumsum(mydf$flowstop)

# clean up vars
mydf$time <- mydf$diff <- mydf$flowstop <- NULL

# find flow length
mydfrle <- rle(mydf$flowid)

# get uniqueid/flowid pairs (unique() is too slow)
mydf <- mydf[!duplicated(mydf$flowid), ]

# append rle and remove separate var
mydf$number_rows <- mydfrle$lengths
rm(mydfrle)

print(Sys.time()-starttime)
# Time difference of 30.39437 secs

数据表：

library(data.table)
starttime <- Sys.time()
res<-setDT(mydf)[, list(number_rows=.N,flow=.GRP),
                 by=.(uniqueid,cumsum(as.numeric(difftime(time,shift(time,1L,type="lag",fill=0))) - 15))][,cumsum:=NULL]
print(Sys.time()-starttime)
# Time difference of 57.08156 secs

dplyr：

library(dplyr)
# convert the time column to POSIXct class so that we can apply the diff function correctly
starttime <- Sys.time()
mydf %>% group_by(uniqueid, flow = 1 + cumsum(c(F, diff(time) != 15))) %>% 
  summarize(num_rows = n())
print(Sys.time()-starttime)
# too long, did not finish after a few minutes

我认为 uniqueid 和时间有序的假设是巨大的，其他解决方案可能能够更好地利用这一点。 order() 很容易做到。

我不确定内存的影响，或者不那么简单的不同数据集的影响。如果内存有问题，应该很容易将其分成块并进行处理。在 Base R 中肯定需要更多代码。

【讨论】：

这不太正确。对 MikeyMike 提供的数据执行 mydf$time[6] = as.POSIXct("2014-05-12 13:30:00") 并重试...
正如 Arun 所说，当它们之间的增量仍为 15 时，此解决方案不会考虑 uniqueids 的差异......
该假设已在 Base R 代码块的第一行中说明（再看一遍，由于其重要性，这是一个放置不当的注释）。如果所有时间段都在数据被提取或记录停止时结束，这是一个合理的假设。速度的提高足以发布，因为无论如何我通常会发现 DT 和 dplyr 快得多。如果由于大小或其他问题无论如何都需要对数据进行分解，则可以很容易地通过 ID 将数据分成块以避免该问题，然后并行处理是提高速度的一种选择。它通常不正确，但它可能会起作用。

【解决方案4】：

拥有有序的“id”和“time”列，我们可以通过创建索引的逻辑向量来构建一个单独的组来操作，只要“id”发生变化或“时间”大于 15 分钟。

与：

id = as.character(mydf$uniqueid)
tm = mydf$time

找到“id”的位置：

id_gr = c(TRUE, id[-1] != id[-length(id)])

和“时间”：

tm_gr = c(0, difftime(tm[-1], tm[-length(tm)], unit = "mins")) > 15

改变并结合它们：

gr = id_gr | tm_gr

显示“id”更改或“时间”> 15 的任何位置。并得到结果：

tab = tabulate(cumsum(gr))  ## basically, the only operation per group -- 'n by group'
data.frame(id = id[gr], flow = seq_along(tab), n = tab)
#  id flow n
#1  a    1 4
#2  a    2 2
#3  b    3 2
#4  b    4 1

在更大的范围内：

set.seed(1821); nid = 1e4         
dat = replicate(nid, as.POSIXct("2016-07-07 12:00:00 EEST") + 
                     cumsum(sample(c(1, 5, 10, 15, 20, 30, 45, 60, 90, 120, 150, 200, 250, 300), sample(5e2:1e3, 1), TRUE)*60),
                simplify = FALSE)
names(dat) = make.unique(rep_len(letters, nid))
dat = data.frame(id = rep(names(dat), lengths(dat)), time = do.call(c, dat))

system.time({
    id = as.character(dat$id); tm = dat$time
    id_gr = c(TRUE, id[-1] != id[-length(id)])
    tm_gr = c(0, difftime(tm[-1], tm[-length(tm)], unit = "mins")) > 15
    gr = id_gr | tm_gr
    tab = tabulate(cumsum(gr))
    ans1 = data.frame(id = id[gr], flow = seq_along(tab), n = tab)
})
# user  system elapsed 
#  1.44    0.19    1.66

为了比较，包括 MikeyMike 的回答：

library(data.table)
dat2 = copy(dat)
system.time({
    ans2 = setDT(dat2)[, list(flow = .GRP, n = .N),
                by = .(id, cumsum(as.numeric(difftime(time, 
                                      shift(time, 1L, type = "lag", fill = 0), 
      unit = "mins")) > 15))][, cumsum := NULL]    
})
# user  system elapsed 
# 3.95    0.22    4.26

identical(as.data.table(ans1), ans2)
#[1] TRUE

【讨论】：

这真是太棒了。迄今为止最快且非常直接的。非常感谢！！
另外一个问题；我想根据原始 df 执行一些计算（df 中有更多列我没有包含在示例中）。您将如何将流编号转换回原始数据集（因此每行都分配了它的 flynr，因此多行将具有相同的 flynr）？
@zoekdestep ：如果我理解正确，你需要像 cbind(mydf, flow = cumsum(gr))? IE。拥有gr 并应用cumsum 会得到nrow(mydf) 向量。
就这么简单 :) 非常感谢，你真的帮了我！