在数据分析中，我们经常面临需要将两个数据集基于多个维度进行关联的场景，其中一个关键维度是时间窗口。例如，我们可能需要将用户的交易记录与该交易发生前一定时间范围内的浏览行为进行匹配。传统的pd.merge函数主要用于等值连接，而pd.merge_asof虽然可以处理近似时间匹配，但它通常只匹配最近的一个条目，并且会“消耗”已匹配的条目，这不适用于需要获取时间窗口内所有相关记录的场景。
为了更好地说明问题，我们使用以下两个DataFrame作为示例：一个记录了交易信息（trade），另一个记录了用户的浏览历史（view）。我们的目标是为每笔交易找出其发生前7天内的所有相关浏览记录（基于相同的person和code），并将这些浏览记录的日期和值以列表形式附加到交易数据中。
import pandas as pd
import janitor # 确保已安装：pip install pyjanitor
# 交易数据DataFrame
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]
})
# 浏览历史DataFrame
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]
})
# 将日期列转换为datetime类型，以便进行时间计算和比较
trade['date'] = pd.to_datetime(trade['date'])
view['date'] = pd.to_datetime(view['date'])1. 使用 janitor.conditional_join 实现高效多条件关联
pyjanitor库提供了一个功能强大的conditional_join函数，它允许我们指定多个非等值连接条件，这在处理时间范围匹配等复杂场景时非常有用，并且通常比纯Pandas的宽泛合并再筛选的方法更高效。
实现步骤：
- 创建辅助时间列： 对于每笔交易，计算其发生日期前7天的日期作为时间窗口的起始点。
-
执行条件连接： 使用conditional_join函数，指定以下连接条件：
- trade的start_date小于等于view的view_dates（确保浏览日期在窗口起始点之后或等于）。
- trade的date大于等于view的view_dates（确保浏览日期在交易日期之前或等于）。
- trade的person等于view的person。
- trade的code等于view的code。
- 重命名和格式化： 重命名view DataFrame中的日期和值列，并将最终的浏览日期格式化为字符串。
- 聚合结果： 使用groupby和agg(list)将每个交易的所有匹配浏览记录聚合成列表。
# 使用janitor.conditional_join进行数据关联
out_janitor = (
trade
.assign(start_date=lambda d: d['date'].sub(pd.DateOffset(days=7))) # 步骤1: 计算时间窗口起始点
.conditional_join(
view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 步骤2: 重命名view列
('start_date', 'view_dates', '<='), # 条件1: 浏览日期 >= 交易日期 - 7天
('date', 'view_dates', '>='), # 条件2: 浏览日期 <= 交易日期
('person', 'person', '=='), # 条件3: person相等
('code', 'code', '=='), # 条件4: code相等
right_columns=['view_dates', 'view_values'] # 保留右侧DataFrame的哪些列
)
.drop(columns='start_date') # 步骤3: 删除辅助列
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 步骤3: 格式化日期
.groupby(list(trade.columns), as_index=False).agg(list) # 步骤4: 按原始trade列分组并聚合
)
print("使用janitor.conditional_join的结果:")
print(out_janitor)优点与注意事项：janitor.conditional_join在内部优化了连接过程，特别是在处理多个非等值条件时，能够避免生成过大的中间DataFrame，从而提高性能。对于大规模数据集和复杂的时间窗口关联，它是更为推荐的选择。
2. 纯 Pandas 解决方案
如果没有pyjanitor库，我们也可以使用纯Pandas实现相同的功能。这种方法通常涉及一个更宽泛的初始合并，然后通过条件筛选来缩小结果范围。
实现步骤：
- 宽泛合并： 首先，基于person和code这两个等值条件，对trade和view进行合并。这将生成一个包含所有可能组合的中间DataFrame。
-
时间窗口筛选： 在合并后的DataFrame上，应用两个时间条件进行筛选：
- 浏览日期（view_dates）必须小于或等于交易日期（date）。
- 浏览日期（view_dates）必须大于或等于交易日期减去7天（date - pd.DateOffset(days=7)）。
- 重命名和格式化： 同样重命名view DataFrame中的日期和值列，并将最终的浏览日期格式化为字符串。
- 聚合结果： 使用groupby和agg(list)将每个交易的所有匹配浏览记录聚合成列表。
# 纯Pandas解决方案进行数据关联
out_pandas = (
trade
.merge(
view.rename(columns={'date': 'view_dates', 'value': 'view_values'}), # 步骤1: 重命名view列
on=['person', 'code'], # 步骤1: 基于person和code进行等值合并
how='left' # 使用左连接以保留所有交易，即使没有匹配的浏览记录
)
.loc[lambda d: (d['date'].gt(d['view_dates'])) & # 步骤2: 浏览日期在交易日期之前
(d['date'].sub(pd.DateOffset(days=7)).le(d['view_dates'])) # 步骤2: 浏览日期在窗口起始点之后
]
.assign(view_dates=lambda d: d['view_dates'].dt.strftime('%Y-%m-%d')) # 步骤3: 格式化日期
.groupby(list(trade.columns), as_index=False).agg(list) # 步骤4: 按原始trade列分组并聚合
)
print("\n纯Pandas解决方案的结果:")
print(out_pandas)优点与注意事项： 纯Pandas方案的缺点在于，merge操作会首先生成一个包含所有person和code组合的笛卡尔积（如果how='inner'或how='left'），即使这些组合在时间上不匹配。对于大型数据集，这可能导致中间DataFrame非常庞大，占用大量内存并影响性能。因此，这种方法更适用于数据量相对较小或对性能要求不那么严格的场景。
3. 总结
本文详细介绍了两种在Pandas中处理多条件和时间窗口数据关联的方法。
- janitor.conditional_join 提供了更灵活和高效的解决方案，特别适用于需要处理非等值连接（如时间范围）和避免生成大规模中间结果的场景。它通过优化内部连接逻辑，能够更有效地处理复杂的条件匹配。
- 纯 Pandas 方案 虽然易于理解和实现，但其性能在大数据集上可能受限于生成中间DataFrame的规模。它通过先进行宽泛合并再筛选的方式实现，可能导致内存消耗较大和计算效率降低。
在实际应用中，建议根据数据集的规模、复杂度和性能要求来选择合适的工具和方法。对于复杂的非等值连接，pyjanitor库是一个值得考虑的强大补充，能够显著提升数据处理的效率和可维护性。
