删除数百万个压缩 CSV 文件中的重复行，同时保留重复行中的一条信息

Question

拥有约 1000 万个 GZipped CSV 文件的集合，每个文件都包含 100-1000 行和 >2000 列。每个文件还包含一个标题。

在每个 CSV 文件中都有两个重要的列，“ID”和“目标”。

我正在尝试删除具有重复“目标”的行，但将要删除的行中的 ID 与不会被删除的行一起保留。

例如

输入：

CSV1
|   ID  |  Target                      |
|-------|------------------------------|
| IX213 | C1=CC(=CC=C1CC(=O)C(=O)O)O   |
| IX412 | CN1C=NC2=C1C(=O)N(C(=O)N2C)C |

CSV2
|   ID  |  Target                      |
|-------|------------------------------|
| BC144 | CN1C=NC2=C1C(=O)N(C(=O)N2C)C |
| BC155 | C(CC(=O)O)C(C(=O)O)N         |

输出：

CSV1*
|   ID         |  Target                      |
|--------------|------------------------------|
| IX213        | C1=CC(=CC=C1CC(=O)C(=O)O)O   |
| IX412; BC144 | CN1C=NC2=C1C(=O)N(C(=O)N2C)C |

CSV2*
|   ID  |  Target                      |
|-------|------------------------------|
| BC155 | C(CC(=O)O)C(C(=O)O)N         |

这对于使用 Pandas (Python) 等的少量文件来说是很简单的，但希望有人可能有更好的方法来处理具有数十亿条目的数百万个文件。

Answer 1

我会编写一个程序，它会遍历一个 gzip 压缩的 csv 文件并写入以下 4 列：target id file row。

对每个文件运行此程序，您将获得约 1000 万个小文件。

假设您不以分布式方式执行此操作，接下来我会将文件合并为一个大文件，并使用 unix 排序实用程序对其进行排序。 （警告您要执行 LC_ALL=C sort foo.txt，因为 C 语言环境更快，并且产生更合理的结果。有关更多信息，请参阅 sort not sorting as expected (space and locale)。）

现在处理该文件并决定保留哪个文件很容易。您可以使用列file row target id is_keep removed_ids 写出一个文件。确保用前导零写出该行，因此您应该写000042 而不是42。 removed_ids 是您从其他文件中删除的那些，如果您保留了这个。 （前导零的数量应该足以满足您最大的文件。即使 asciibetical 顺序匹配数字顺序。）

再次对该文件进行排序，然后将其分解为每个文件决定的文件。

鉴于原始 gzip 文件以及该文件要保留哪些行，以及如果您保留它要存储哪些 ID，处理您的原始文件以删除/保留行并记录您删除的内容很容易。我强烈建议进行完整性检查以验证目标/ID/行是否全部匹配。除非完整性检查通过，否则不要删除原件。

---

如果您是分布式处理的忠实拥护者，那么从排序到 map-reduce 的转换非常简单。如果您有该设置的基础架构，则不妨使用它。如果没有，我会建议这种排序文件方法，仅使用并行化处理第一个/最后一个的所有单个文件。

Answer 2

尽管数据量可能看起来很庞大，但我认为如果您保留所需的适量数据，您可以按顺序迭代所有文件。例如，您可以跟踪与具有此类目标的第一个 ID 和 ID 别名 关系的唯一目标的关系（例如，ID IX412 对应于 BC144）。这样，您的解决方案可能如下所示：

import csv

filenames = [...]
target_ids = {}
aliases = {}

for filename in filenames:
    with open(filename, 'r') as file_in:
        reader = csv.DictReader(file_in)
        for row in reader:
            if row['Target'] in target_ids:
                aliases[row['ID']] = target_ids[row['Target']]
                remove_row(row)  # Do whatever you may require
            else:
                target_ids[row['Target']] = row['ID']

请注意，拥有 10M 键值对的 dict 非常容易处理。

如果这仍然不适合内存，您可以使用 shelve 而不是 dicts，以便将相应的数据存储在 HDD 中。你可以这样做：

import csv
import shelve

filenames = [...]

with shelve.open('target_ids') as target_ids, shelve.open('aliases') as aliases:
    for filename in filenames:
        with open(filename, 'r') as file_in:
            reader = csv.DictReader(file_in)
            for row in reader:
                if row['Target'] in target_ids:
                    aliases[row['ID']] = target_ids[row['Target']]
                    remove_row(row)  # Do whatever you may require
                else:
                    target_ids[row['Target']] = row['ID']

shelve 的缺点在于常规 dict 的速度。