基于Python优化多厂商用户时间戳重叠处理方案

针对千万级数据处理慢的问题，核心优化思路是替换逐行循环为向量化计算、利用高效分组排序，必要时用分块并行处理，以下是具体实现：

1. 预处理：排序分组减少重复计算 #

先按用户ID（关联多厂商账户的核心标识）和起始时间排序，确保同用户的时间戳按顺序排列，后续重叠判断只需对比相邻行：

import pandas as pd

# 假设数据结构：user_id, vendor_id, start_time, end_time
df = pd.read_csv('your_data.csv', parse_dates=['start_time', 'end_time'])
# 按用户分组，组内按起始时间排序
df_sorted = df.sort_values(['user_id', 'start_time']).reset_index(drop=True)

2. 向量化标记两种重叠类型 #

用Pandas的shift函数获取同用户内的上一行时间戳，避免循环判断：

标记完全覆盖 #

判断当前行时间戳是否被上一行完全包含，或完全包含上一行：

# 同用户内取上一行的起始、结束时间
df_sorted['prev_start'] = df_sorted.groupby('user_id')['start_time'].shift(1)
df_sorted['prev_end'] = df_sorted.groupby('user_id')['end_time'].shift(1)

# 标记完全覆盖的行
df_sorted['is_full_cover'] = (
    (df_sorted['start_time'] >= df_sorted['prev_start']) & (df_sorted['end_time'] <= df_sorted['prev_end'])
) | (
    (df_sorted['prev_start'] >= df_sorted['start_time']) & (df_sorted['prev_end'] <= df_sorted['end_time'])
)

标记部分重叠 #

判断存在时间交集但不满足完全覆盖的情况：

df_sorted['is_partial_overlap'] = (
    (df_sorted['start_time'] < df_sorted['prev_end']) & (df_sorted['end_time'] > df_sorted['prev_start'])
) & ~df_sorted['is_full_cover']

3. 合并/去重重叠区间 #

对完全覆盖的区间保留范围更大的，部分重叠的合并为连续区间：

def process_user_group(group):
    # 过滤被完全覆盖的行（保留覆盖范围大的）
    filtered = group[~group['is_full_cover']].copy()
    
    # 合并部分重叠的区间
    merged = []
    for _, row in filtered.iterrows():
        if not merged:
            merged.append(row.to_dict())
        else:
            last = merged[-1]
            if row['start_time'] <= last['end_time']:
                # 合并区间，记录关联的厂商ID
                last['end_time'] = max(last['end_time'], row['end_time'])
                last['vendor_ids'] = f"{last['vendor_id']},{row['vendor_id']}"
            else:
                merged.append(row.to_dict())
    return pd.DataFrame(merged)

# 按用户分组处理
result_df = df_sorted.groupby('user_id').apply(process_user_group).reset_index(drop=True)

4. 超大数据量的进阶优化（内存不足时） #

如果数据量超过内存，用Dask做分块并行处理，自动拆分任务到多个核心：

import dask.dataframe as dd

# 用Dask读取数据，自动分块
ddf = dd.read_csv('your_data.csv', parse_dates=['start_time', 'end_time'])
ddf_sorted = ddf.sort_values(['user_id', 'start_time'])

# 重复上述标记逻辑（Dask语法与Pandas几乎一致）
ddf_sorted['prev_start'] = ddf_sorted.groupby('user_id')['start_time'].shift(1)
ddf_sorted['prev_end'] = ddf_sorted.groupby('user_id')['end_time'].shift(1)
ddf_sorted['is_full_cover'] = (
    (ddf_sorted['start_time'] >= ddf_sorted['prev_start']) & (ddf_sorted['end_time'] <= ddf_sorted['prev_end'])
) | (
    (ddf_sorted['prev_start'] >= ddf_sorted['start_time']) & (ddf_sorted['prev_end'] <= ddf_sorted['end_time'])
)
ddf_sorted['is_partial_overlap'] = (
    (ddf_sorted['start_time'] < ddf_sorted['prev_end']) & (ddf_sorted['end_time'] > ddf_sorted['prev_start'])
) & ~ddf_sorted['is_full_cover']

# 并行处理分组计算
result_ddf = ddf_sorted.groupby('user_id').apply(process_user_group, meta=result_df.dtypes).compute()

性能提升关键 #

• 全程用Pandas/Dask的向量化操作，避免Python原生循环（逐行处理是千万级数据慢的核心原因）
• 提前解析时间戳为datetime类型，避免字符串比较的开销
• 用Parquet替代CSV存储，读写速度提升数倍：df.to_parquet('processed_data.parquet')
• 若无用户关联标识，可尝试用intervaltree库构建区间索引，快速查询重叠区间，但需注意内存占用

内容的提问来源于stack exchange，提问作者Mr.Oz