R中合并关联数据行的最优方法：合并间隔≤10分钟的Dual记录

我来帮你梳理下针对大文件合并Dual记录的优化思路，先把问题拆解清楚：你的需求是将Bolus Type为Dual (normal part only)和对应的Dual (square part only)行合并，要求两行时间间隔不超过10分钟，但现有代码处理>30MB的文件时速度太慢，下面给你针对性的优化方案。

先明确你的示例数据 #

现有代码慢的核心原因 #

大概率是逐行迭代处理（比如用for循环遍历每一行），没有利用数据处理库的向量化操作，导致CPU利用率极低；另外如果直接把整个大文件加载到内存，也可能因为内存交换拖慢速度。

高效优化方案（以Pandas为主，适配大文件） #

下面从内存友好、计算高效两个维度给出具体实现：

1. 分块读取大文件（避免内存溢出）

如果文件太大无法一次性加载，用pandas.read_csv的chunksize参数分块处理，把大文件拆成小批量处理：

import pandas as pd

chunk_size = 10000  # 可根据你的内存容量调整块大小
output_chunks = []

# 分块读取并处理
for chunk in pd.read_csv('your_data.csv', chunksize=chunk_size):
    merged_chunk = process_dual_records(chunk)
    output_chunks.append(merged_chunk)

# 合并所有块的结果并保存
final_result = pd.concat(output_chunks, ignore_index=True)
final_result.to_csv('merged_data.csv', index=False)

2. 向量化合并逻辑（替代逐行循环）

用Pandas的移位、布尔筛选等向量化操作替代逐行判断，这是提升速度的关键：

def process_dual_records(df):
    # 标记Dual类型的两行
    df['is_normal_dual'] = df['Bolus Type'] == 'Dual (normal part only)'
    df['is_square_dual'] = df['Bolus Type'] == 'Dual (square part only)'
    
    # 移位获取下一行的时间和类型，用于匹配
    df['next_time'] = df['Time'].shift(-1)
    df['next_type'] = df['Bolus Type'].shift(-1)
    
    # 筛选符合条件的normal dual行：下一行是square dual，且时间差≤10分钟
    valid_normal = df[
        df['is_normal_dual'] & 
        (df['next_type'] == 'Dual (square part only)') &
        ((df['next_time'] - df['Time']) <= 10)
    ].copy()
    
    # 获取对应的square行（就是normal行的下一行）
    valid_square = df.loc[valid_normal.index + 1].copy()
    
    # 合并两行数据：保留normal行的时间，合并Volume，类型改为Dual
    merged_dual = pd.DataFrame({
        'Time': valid_normal['Time'],
        'Bolus Type': 'Dual',
        'Bolus Volume': valid_normal['Bolus Volume'] + valid_square['Bolus Volume']
    })
    
    # 筛选不需要合并的行：非Dual行，或者不符合合并条件的Dual行
    exclude_indices = pd.concat([valid_normal.index, valid_normal.index + 1])
    non_merged = df[~df.index.isin(exclude_indices)].drop(
        columns=['is_normal_dual', 'is_square_dual', 'next_time', 'next_type']
    )
    
    # 合并结果并按时间排序
    return pd.concat([non_merged, merged_dual]).sort_values('Time').reset_index(drop=True)

3. 额外优化：压缩数据类型

加载文件时指定更小的数据类型，减少内存占用，进一步提升处理速度：

# 定义压缩后的数据类型
dtypes = {
    'Time': 'float32',
    'Bolus Type': 'category',  # 类型是有限枚举，用category省内存
    'Bolus Volume': 'int16'
}

# 分块读取时指定dtypes
for chunk in pd.read_csv('your_data.csv', chunksize=chunk_size, dtype=dtypes):
    merged_chunk = process_dual_records(chunk)
    output_chunks.append(merged_chunk)

4. 极端大文件：用Dask并行处理

如果文件大到Pandas分块都吃力，可以用Dask（并行处理库），语法和Pandas几乎一致，能自动分块并行计算：

import dask.dataframe as dd

df = dd.read_csv('your_data.csv', dtype=dtypes)
# 自动并行处理每个数据块
processed_df = df.map_partitions(process_dual_records)
# 保存为单个文件
processed_df.to_csv('merged_data.csv', single_file=True)

验证结果 #

用你的示例数据测试，合并后会得到如下结果，完全符合需求：

内容的提问来源于stack exchange，提问作者Simon Brackley