问题场景与挑战
在数据分析中，我们经常需要关联两个数据集，其中一个数据集的记录（例如交易记录）需要匹配另一个数据集（例如浏览历史）中发生在特定时间窗口内的所有相关记录。考虑以下两个pandas dataframe：
-
trade DataFrame：包含交易日期、人员ID、商品代码和交易价值。
import pandas as pd import janitor # 用于方案一 trade = pd.DataFrame({'date': ['2019-08-31', '2019-09-01', '2019-09-04'], 'person': [1, 1, 2], 'code': [123, 123, 456], 'value1': [1, 2, 3]}) -
view DataFrame：包含浏览日期、人员ID、商品代码和浏览价值。
view = pd.DataFrame({'date': ['2019-08-29', '2019-08-29', '2019-08-30', '2019-08-31', '2019-09-01', '2019-09-01', '2019-09-01', '2019-09-02', '2019-09-03'], 'person': [1, 1, 1, 2, 1, 2, 2, 1, 2], 'code': [123, 456, 123, 456, 123, 123, 456, 123, 456], 'value': [1, 2, 3, 4, 5, 6, 7, 8, 9]})
我们的目标是为每一笔交易，找出其发生前7天内（含交易当天）所有匹配person和code的浏览记录，并将这些浏览记录的日期和价值聚合为列表。
直接使用Pandas的merge_asof函数在这种场景下存在局限性。merge_asof设计用于“最近匹配”，它会为左DataFrame的每一行找到右DataFrame中最近的匹配行，且通常只匹配一次。这不符合我们“获取所有在时间窗口内的匹配项”的需求。例如，merge_asof可能会将view中的[1, 3]分配给第一笔交易，但对于第二笔交易，它可能只会分配[5]，而忽略了同样在时间窗口内的[1, 3]。
方案一：使用pyjanitor.conditional_join进行高效多条件连接
pyjanitor库提供了一个强大的conditional_join函数，它允许基于多个自定义条件进行连接，包括非等值条件（如时间范围）。这使得它成为解决此类问题的理想选择，尤其是在处理大型数据集时，其效率通常高于纯Pandas的通用合并再筛选方案。
实现步骤
- 日期列转换：确保所有涉及日期比较的列都是Pandas的datetime类型。
- 创建辅助时间列：在trade DataFrame中，为每笔交易计算一个“开始日期”，即交易日期减去7天。这将作为时间窗口的下限。
-
执行条件连接：使用conditional_join连接trade和view DataFrame。连接条件包括：
- view的日期在trade的start_date之后或等于。
- view的日期在trade的date之前或等于。
- person列相等。
- code列相等。
- 数据聚合：连接完成后，结果DataFrame中每行将是trade记录与一个匹配的view记录的组合。我们需要根据原始trade的列进行分组，然后将匹配到的view_dates和view_values聚合成列表。
- 格式化输出：将聚合后的日期列表转换回字符串格式。
代码示例
# 确保日期列为datetime类型
trade['date'] = pd.to_datetime(trade['date'])
view['date'] = pd.to_datetime(view['date'])
# 方案一：使用janitor.conditional_join
out_janitor = (trade
.assign(start_date=lambda d: d['date'].sub(pd.DateOffset(days=7))) # 计算时间窗口的开始日期
.conditional_join(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view的列以避免冲突
# 定义连接条件：(左列, 右列, 操作符)
('start_date', 'view_dates', '<='), # view_dates >= start_date
('date', 'view_dates', '>='), # view_dates <= trade_date
('person', 'person', '=='), # person相等
('code', 'code', '=='), # code相等
right_columns=['view_dates', 'view_values'] # 只保留view中需要的列
)
.drop(columns='start_date') # 移除辅助列
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade.columns), as_index=False).agg(list) # 按原始trade列分组并聚合为列表
)
print("--- 使用 janitor.conditional_join 的结果 ---")
print(out_janitor)优势分析
conditional_join在内部实现了优化的算法来处理非等值连接，特别是涉及到范围查询时，其性能通常优于先进行笛卡尔积式合并再筛选的纯Pandas方法，尤其适用于大数据集。
方案二：纯Pandas的通用合并与筛选
虽然conditional_join更高效，但也可以纯粹使用Pandas的merge和loc进行筛选来实现相同的功能。这种方法在概念上更直接，但可能在性能上有所牺牲，因为它首先会生成一个更大的中间DataFrame。
实现步骤
- 日期列转换：同方案一，确保日期列为datetime类型。
- 通用合并：使用merge函数基于person和code列进行内连接。这将生成trade和view中所有person和code匹配的组合。
- 时间窗口筛选：使用loc根据时间条件（view_dates在trade_date前7天内）对合并后的结果进行筛选。
- 数据聚合：同方案一，根据原始trade的列进行分组，然后将匹配到的view_dates和view_values聚合成列表。
- 格式化输出：将聚合后的日期列表转换回字符串格式。
代码示例
# 确保日期列为datetime类型
trade['date'] = pd.to_datetime(trade['date'])
view['date'] = pd.to_datetime(view['date'])
# 方案二：纯Pandas解决方案
out_pandas = (trade
.merge(view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 重命名view的列
on=['person', 'code']) # 基于person和code进行合并
.loc[lambda d: d['date'].gt(d['view_dates']) & # 交易日期必须晚于浏览日期
d['date'].sub(pd.DateOffset(days=7)).le(d['view_dates']) # 浏览日期必须在交易日期前7天内
]
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 格式化日期为字符串
.groupby(list(trade.columns), as_index=False).agg(list) # 按原始trade列分组并聚合为列表
)
print("\n--- 纯Pandas解决方案的结果 ---")
print(out_pandas)局限性分析
纯Pandas方案首先会生成一个包含所有person和code匹配组合的中间DataFrame。如果trade和view中存在大量相同person和code的记录，这个中间DataFrame可能会非常大，导致内存消耗增加和计算时间延长。随后进行的时间条件筛选会减少数据量，但前期合并的开销是不可避免的。
总结与最佳实践
两种方法都能达到预期的结果：
date person code value1 view_dates view_values 0 2019-08-31 1 123 1 [2019-08-29, 2019-08-30] [1, 3] 1 2019-09-01 1 123 2 [2019-08-29, 2019-08-30, 2019-09-01] [1, 3, 5] 2 2019-09-04 2 456 3 [2019-08-31, 2019-09-01, 2019-09-03] [4, 7, 9]
- 对于小型数据集或对性能要求不高的场景：纯Pandas解决方案简单直观，易于理解和实现。
- 对于大型数据集或对性能有严格要求的场景：强烈推荐使用pyjanitor.conditional_join。它通过更优化的算法处理非等值连接，能够显著提高效率并减少内存消耗。
注意事项：
- 日期类型一致性：在进行任何日期比较或计算之前，务必确保所有日期列都已正确转换为Pandas datetime类型。
- 时间窗口定义：根据具体业务需求精确定义时间窗口（例如，pd.DateOffset(days=7)表示7天，可以根据需要调整为hours、minutes等）。
- 列名冲突：在合并前，如果两个DataFrame中有相同名称但含义不同的列，请务必重命名以避免混淆。在示例中，我们重命名了view中的date和value列。
- 安装pyjanitor：如果选择使用pyjanitor，请确保已通过pip install pyjanitor安装该库。
通过上述两种方法，我们可以灵活地在Pandas中处理复杂的基于多条件和时间窗口的数据关联任务，选择最适合当前项目需求和数据规模的方案。
