大数据集下高效关联A类与±5分钟内B类传感器数据并汇总

问题需求 #

在大型传感器数据集中，需按个体为每个datatype为"B"的观测值，匹配其datetime±5分钟内的"A"类型观测值，计算这些"A"值的均值与数量，并将该信息关联到对应"B"记录。现有嵌套循环方案在数十万行数据中运行极慢，耗时数小时，需更高效的实现方法。

示例数据 #

Individual <- c(rep("Anna",7),rep("Mark",8))
datetime <- as.POSIXct(c("2021-11-18 07:02:12","2021-11-18 07:10:25","2021-11-18 07:22:18","2021-11-18 07:24:04","2021-11-18 07:27:43","2021-11-18 07:29:01","2021-11-18 07:50:01","2021-11-21 12:19:28","2021-11-21 10:55:58","2021-11-21 11:14:22","2021-11-21 11:18:05","2021-11-21 11:22:01","2021-11-21 11:26:09","2021-11-21 11:27:38","2021-11-21 12:15:50"), format = "%Y-%m-%d %H:%M:%S")
datatype <- as.factor(c("A","A","A","B","A","A","A","A","A","B","A","B","A","A","A")) #type of sensordata
value <- c(7.85,6.54,7.82,5.43,7.34,8.93,2.54,5.62,7.84,5.56,8.95,6.85,5.88,4.95,8.46) #value of the sensor

df<-data.frame(Individual, datetime, datatype, value)

原低效嵌套循环代码 #

library(dplyr)
library(plyr)

df_A<-df %>% filter(datatype == "A")
df_B<-df %>% filter(datatype == "B")

df_list <- list() 
name_list <- list() 
ind<-unique(df_B$Individual)

for (i in 1:length(ind)) { # loop over all individuals in the data
  for (j in 1:nrow(df_B[df_B$Individual==paste(ind[i]),])) { #loop over each observation in the data per individual
    row_number<-which(near(df_B[df_B$Individual==paste(ind[i]),]$datetime[j], 
                      df_A[df_A$Individual==paste(ind[i]),]$datetime, tol=5)) #find observations of "A" within +- 5 min an observation of "B"
    #summarize results in a dataframe format
    df1 <- df_B[df_B$Individual==paste(ind[i]),][j,]
    mean_for_A <- mean(df_A[df_A$Individual==paste(ind[i]),][row_number,]$value) # calc. mean of "A"
    number_of_A <- length(df_A[df_A$Individual==paste(ind[i]),][row_number,]$value) #calc. number of "A"
    df2<-cbind(df1, mean_for_A, number_of_A) #combine into dataframe 

    #store dataframe in list
    df_list[[j]]<-df2 #store dataframe of observations of "A" for each "B"
    name_list[[i]]<-ldply(df_list) #store completed dataframes per individual
    
  }
}

ldply(name_list) #final product as I want it to be

高效解决方案 #

方法一：使用data.table的非等连接 #

data.table的非等连接是处理这类时间窗口匹配的高效方案，能避免嵌套循环的冗余计算，大幅提升速度。

library(data.table)

# 转换为data.table格式
setDT(df)

# 提取A、B类数据
dt_A <- df[datatype == "A"]
dt_B <- df[datatype == "B"]

# 为B类观测计算时间窗口：datetime ±5分钟
dt_B[, `:=`(start_time = datetime - 5*60, end_time = datetime + 5*60)]

# 非等连接匹配+分组统计
result <- dt_B[dt_A, on = .(Individual, start_time <= datetime, end_time >= datetime), 
               .(Individual, B_datetime = x.datetime, B_value = x.value, A_value = i.value), 
               allow.cartesian = TRUE][, 
                 .(mean_for_A = mean(A_value), number_of_A = .N), 
                 by = .(Individual, B_datetime, B_value)]

# 整理成目标格式
setcolorder(result, c("Individual", "B_datetime", "B_value", "mean_for_A", "number_of_A"))
setnames(result, c("B_datetime", "B_value"), c("datetime", "value"))
result

方法二：使用fuzzyjoin包的模糊连接 #

fuzzyjoin提供了更直观的模糊连接语法，适合习惯dplyr风格的用户，性能远优于嵌套循环。

library(dplyr)
library(fuzzyjoin)

# 筛选A、B类数据
df_A <- df %>% filter(datatype == "A")
df_B <- df %>% filter(datatype == "B")

# 模糊连接匹配时间窗口+分组统计
result <- df_B %>%
  fuzzy_inner_join(df_A, 
                   by = c("Individual" = "Individual", 
                          "datetime" = "datetime"),
                   match_fun = list(`==`, function(x, y) y >= x - 5*60 & y <= x + 5*60)) %>%
  group_by(Individual, datetime.x, value.x) %>%
  summarise(mean_for_A = mean(value.y),
            number_of_A = n(),
            .groups = "drop") %>%
  rename(datetime = datetime.x, value = value.x)

result

内容的提问来源于stack exchange，提问作者Dave Larsson