两个日期之间合并大型 Pandas 数据框的有效方法

Question

我知道有很多类似于这个问题，但是我似乎找不到相关的答案。 假设我有两个数据框如下：

df1 = pd.DataFrame(
    {
        "end": [
            "2019-08-31",
            "2019-08-28",
            "2019-09-09",
            "2019-09-08",
            "2019-09-14",
            "2019-09-14",
        ],
        "start": [
            "2019-08-27",
            "2019-08-22",
            "2019-08-04",
            "2019-09-02",
            "2019-09-06",
            "2019-09-10",
        ],
        "id": [1234, 8679, 8679, 1234, 1234, 8679],
    }
)

df2 = pd.DataFrame(
    {
        "timestamp": [
            "2019-08-30 10:00",
            "2019-08-28 10:00",
            "2019-08-27 10:30",
            "2019-08-07 12:00",
            "2019-09-12 10:00",
            "2019-09-11 14:00",
            "2019-08-29 18:00",
        ],
        "id": [1234, 1234, 8679, 1234, 8679, 8679, 1234],
        "val": ["AAAB", "ABBA", "CXXC", "BBAA", "XCXC", "CCXX", "BAAB"],
    }
)

df1["end"] = pd.to_datetime(df1["end"])
df1["start"] = pd.to_datetime(df1["start"])

df2["timestamp"] = pd.to_datetime(df2["timestamp"])

df1.sort_values(by=["end"], inplace=True)
df2.sort_values(by="timestamp", inplace=True)

Resulted as:

 end       start    id
0  2019-08-31  2019-08-27  1234
1  2019-08-28  2019-08-22  8679
2  2019-09-09  2019-08-04  8679
3  2019-09-08  2019-09-02  1234
4  2019-09-14  2019-09-06  1234
5  2019-09-14  2019-09-10  8679

 timestamp    id   val
0  2019-08-30 10:00  1234  AAAB
1  2019-08-28 10:00  1234  ABBA
2  2019-08-27 10:30  8679  CXXC
3  2019-08-07 12:00  1234  BBAA
4  2019-09-12 10:00  8679  XCXC
5  2019-09-11 14:00  8679  CCXX
6  2019-08-29 18:00  1234  BAAB

将 df1 按照 ID 合并，以使时间戳位于 start 和 end 之间的常见方法是按照 ID 或虚拟变量进行合并并筛选：

merged_df = pd.merge(df1, df2, how="left", on="id")
merged_df = merged_df.loc[
    (merged_df["timestamp"] >= merged_df["start"])
    & (merged_df["timestamp"] <= merged_df["end"])
]

我得到了我想要的输出：

           end       start    id         timestamp   val
0   2019-08-31  2019-08-27  1234  2019-08-30 10:00  AAAB
1   2019-08-31  2019-08-27  1234  2019-08-28 10:00  ABBA
3   2019-08-31  2019-08-27  1234  2019-08-29 18:00  BAAB
4   2019-08-28  2019-08-22  8679  2019-08-27 10:30  CXXC
7   2019-09-09  2019-08-04  8679  2019-08-27 10:30  CXXC
19  2019-09-14  2019-09-10  8679  2019-09-12 10:00  XCXC
20  2019-09-14  2019-09-10  8679  2019-09-11 14:00  CCXX

我的问题：我需要做相同的合并并获得相同的结果，但df1有200K行，而df2有600K行。

我迄今为止尝试过的：

• 像上面那样进行经典合并和过滤将会失败，因为初始合并将创建一个巨大的数据框，会使内存超载。

• 我还尝试了pandasql方法，但这导致我的16GB RAM PC卡住了。

• 我尝试了在左连接、右连接和外连接的三个步骤中使用merge_asof，如此处所述，但我运行了一些测试，似乎它总是返回df2中最多2条记录到df1的单行。

任何好的建议都将不胜感激！

- Niv Cohen

---

由于您只关心特定子集而不是整个数据框，所以是否尝试先在df1和df2上进行过滤，然后使用左合并/连接/其他操作？如果您的范围不是整个数据帧，我想它会起作用。我猜如果您只想要子集，那么您真的不需要合并整个数据帧。 - tchar

区间是否重叠？区间索引应该可以解决，但是如果有重叠的索引，可能会有些棘手。 - sammywemmy

还有一件事，df2 中有重复项吗？ - sammywemmy

在df2中没有重复项，但是df2中的多个事件可以匹配df1中的单个记录。如您所见，输出示例中的索引0、1、3。 - Niv Cohen

为什么不将其转储到SQL中并运行呢？或者使用Python的pypolar库，或者R的data.table（它们支持非等连接，这也相当高效）。 - sammywemmy

Answer 1

也许你可以使用 groupby 函数，并使用 pd.IntervalIndex 找到匹配的日期范围，这样就不需要进行 merge 操作了。

def func():
    for x, y in df2.groupby("id"):
        tmp = df1.loc[df1["id"].eq(x)]
        tmp.index = pd.IntervalIndex.from_arrays(tmp['start'], tmp['end'], closed='both')
        y[["start", "end"]] = tmp.loc[y.timestamp, ["start", "end"]].to_numpy()
        yield y

print (pd.concat(func()).sort_index())

            timestamp    id   val      start        end
0 2019-08-30 10:00:00  1234  AAAB 2019-08-27 2019-08-31
1 2019-08-28 10:00:00  1234  ABBA 2019-08-27 2019-08-31
2 2019-08-07 10:30:00  8679  CXXC 2019-08-04 2019-09-09
3 2019-08-27 12:00:00  1234  BBAA 2019-08-27 2019-08-31
4 2019-09-12 10:00:00  8679  XCXC 2019-09-10 2019-09-14
5 2019-09-11 14:00:00  8679  CCXX 2019-09-10 2019-09-14
6 2019-08-29 18:00:00  1234  BAAB 2019-08-27 2019-08-31

- Henry Yik

---

谢谢，这似乎是一个不错的方法，但我似乎无法在我的“真实”数据上使其工作。我一直在这行代码中遇到一个常见的“键错误”：y[["start", "end"]]。 - Niv Cohen

Answer 2

我与niv-dudovitch和david-arenburg一起处理了这个问题，以下是我们的研究成果，希望对你们中的一些人有所帮助... 核心思想是通过创建基于数据子集的数据框列表来防止内存中对象的增长。

第一个版本没有多进程。

import pandas as pd

unk = df1.id.unique()
j = [None] * len(unk)
k = 0

df1.set_index('id', inplace = True)
df2.set_index('id', inplace = True)

for i in unk:
    tmp = df1.loc[df1.index.isin([i])].join(df2.loc[df2.index.isin([i])], how='left')  
    j[k] = tmp.loc[tmp['timestamp'].between(tmp['start'], tmp['end'])]
    k += 1
    
res = pd.concat(j)
res

使用多进程

在我们的实际案例中，我们有两个大型数据框，df2大约有300万行，而df1略高于110K。输出结果约为20M行。

import multiprocessing as mp
import itertools
import concurrent
from concurrent.futures import ProcessPoolExecutor
import time
import pandas as pd
from itertools import repeat


def get_val_between(ids, df1, df2):
    """
    Locate all values between 2 dates by id
    Args:
        - ids (list): list of ids

    Returns:
        - concat list of dataframes
    """

    j = [None] * len(ids)
    k = 0

    for i in ids:
        tmp = df1.loc[df1.index.isin([i])].join(
            df2.loc[df2.index.isin([i])], how="left"
        )
        tmp = tmp.loc[tmp["timestamp"].between(tmp["start"], tmp["end"])]

        # add to list in location k

        j[k] = tmp
        k += 1
    # keep only not None dfs in j

    j = [i for i in j if i is not None]
    if len(j) > 0:
        return pd.concat(j)
    else:
        return None


def grouper(n, iterable, fillvalue=None):
    """grouper(3, 'ABCDEFG', 'x') --> ABC DEF Gxx"""

    args = [iter(iterable)] * n
    return itertools.zip_longest(fillvalue=fillvalue, *args)


def main():

    df1.reset_index(inplace=True, drop=True)
    df2.reset_index(inplace=True, drop=True)
    id_lst = df1.id.unique()
    iter_ids = grouper(10, list(id_lst))

    df1.set_index("id", inplace=True)
    df2.set_index("id", inplace=True)

    # set multi-processes

    executor = concurrent.futures.ProcessPoolExecutor(20)
    result_futures = executor.map(get_val_between, iter_ids, repeat(df1), repeat(df2))
    concurrent.futures.as_completed(result_futures)
    result_concat = pd.concat(result_futures)

    print(result_concat)


if __name__ == "__main__":
    main()

预期的结果如下：

           end      start           timestamp   val
id                                                  
8679 2019-08-28 2019-08-22 2019-08-27 10:30:00  CXXC
8679 2019-09-09 2019-08-04 2019-08-27 10:30:00  CXXC
8679 2019-09-14 2019-09-10 2019-09-11 14:00:00  CCXX
8679 2019-09-14 2019-09-10 2019-09-12 10:00:00  XCXC
1234 2019-08-31 2019-08-27 2019-08-28 10:00:00  ABBA
1234 2019-08-31 2019-08-27 2019-08-29 18:00:00  BAAB
1234 2019-08-31 2019-08-27 2019-08-30 10:00:00  AAAB

作为一个输出2000万行数据的基准测试，多进程方法速度快了10倍。

- Niv Cohen

---

如果你有一些我可以玩耍的样本数据，我会非常感激。 - sammywemmy