有效地比较超大文件的内容

Question

我需要快速比较两个不同格式的文件，但我不知道该怎么做。如果有人能指出我正确的方向，我将不胜感激。

我正在使用 CentOS 6，我最熟悉 Python（Python 2 和 Python 3 都可用）。

---

问题

我正在寻找比较两个大文件的内容（快速）。不幸的是，这些文件的内容不同；在比较它们之前，我需要修改其中的内容。它们组织得不是特别好，所以我不能线性地向下移动并逐行比较。

以下是文件示例：

文件 1 文件 2 作业、时间作业、开始、结束 0123,3-00:00:00 0123,2016-01-01T00:00:00,2016-01-04T00:00:00 1111,05:30:00 1111,2016-01-01T00:00:00,2016-01-01T05:30:00 0000,00:00:05 9090.abc,2016-01-01T12:00:00,2016-01-01T22:00:00 9090.abc,10:00:00 0000,2015-06-01T00:00:00,2015-06-01T00:00:05 ……

我想比较具有相同“工作”字段的行的内容，如下所示：

工作文件 1 内容 文件 2 内容 0123 3-00:00:00 2016-01-01T00:00:00,2016-01-04T00:00:00 1111 05:30:00 2016-01-01T00:00:00,2016-01-01T05:30:00 0000 00:00:05 2015-06-01T00:00:00,2015-06-01T00:00:05 9090.abc 10:00:00 2016-01-01T12:00:00,2016-01-01T22:00:00 …………

我将对 文件 1 内容 和 文件 2 内容 执行计算并比较两者（针对每一行）。

最有效的方法是什么（匹配行）？

---

当前系统会针对另一个文件中的每一行完整地遍历一个文件（直到找到匹配项）。此过程可能需要数小时才能完成，并且文件总是在增长。我希望使比较它们的过程尽可能高效，但即使是性能上的微小改进也会产生巨大的影响。

感谢所有帮助。

谢谢！

Answer 1

如果您能找到利用哈希表的方法，您的任务将从 O(N^2) 变为 O(N)。实施将取决于您的文件有多大以及文件 2 中是否有重复的作业 ID。我假设您没有任何重复项。如果您可以将文件 2 放入内存中，只需将其加载到以作业为索引的 pandas 中即可。如果您无法将文件 2 放入内存中，您至少可以构建一个 {Job #: row # in file 2} 的字典。无论哪种方式，找到匹配项都应该更快。

Answer 2

我能想到的最有效的方法是使用每个现代 Linux 系统都应该具备的一些标准 UNIX 工具。我知道这不是一个 python 解决方案，但你使用 python 的决心似乎主要建立在你对该语言的了解之上，而不是任何外部约束。鉴于此任务使用 UNIX 工具非常简单，我将在此处概述该解决方案。

您要做的是标准数据库样式的连接，您可以在其中查找共享一列的两个表中的信息。为此，必须对文件进行排序，但 UNIX sort 对此使用了一种有效的算法，您不会绕过排序或将文件复制到暗示某种排序的数据结构中。

用于演示的长版

tail -n+2 file1.csv | LC_ALL=C sort -t , -k 1  > file1.sorted.csv
tail -n+2 file2.csv | LC_ALL=C sort -t , -k 1  > file2.sorted.csv
join -a 1 -a 2 -t , -1 1 -2 1 file1.sorted.csv file2.sorted.csv \
   > joined.csv

tail -n+2 切断包含标题的文件的第一行。 -t , 部分是设置逗号作为列分隔符，-k 1 表示按第一列排序。 -1 1 -2 1 表示“使用第一个文件的第一列和第二个文件的第一列作为两个文件的共享列”。 -a 1 -a 2 表示“还从文件 1 和文件 2 输出行，在另一个文件中找不到匹配的行。这与数据库术语中的“完全外部连接”有关。请参阅 this SO question 和其他 LC_ALL=C

如果您想避免保存临时文件，可以使用 bash 的“进程替换”<( ... )即时排序

join -a 1 -a 2 -t , -1 1 -2 1 \
    <( tail -n+2 file1.csv | LC_ALL=C sort -t , -k 1 ) \
    <( tail -n+2 file2.csv | LC_ALL=C sort -t , -k 1 ) \
    > joined.csv

注意sort 支持多核（参见man sort 中的--parallel）在多台机器上，将它们发回并合并排序的部分，请参阅this blog-post。

Answer 3

这是将文件 2 格式转换为文件 1 格式的简单实用程序（我希望我理解问题正确，使用了 python 2） 例如将代码保存到文件util1.py

import time
import sys

if __name__ == '__main__':
    if  len(sys.argv) < 2:
        print 'Err need filename'
        sys.exit()
    with open(sys.argv[1], 'r') as f:
        line = f.next()
        for line in f:
            jb, start, end =  line.rstrip().split(',')
            dt_format ='%Y-%m-%dT%H:%M:%S'
            start = time.strptime(start, dt_format)
            end = time.strptime(end, dt_format)
            delt = time.mktime(end) - time.mktime(start)
            m, s = divmod(delt, 60)
            h, m = divmod(m, 60)
            d, h = divmod(h, 24)
            if d !=0:
                print '{0},{1:d}-{2:02d}:{3:02d}:{4:02d}'.format(jb, int(d), int(h), int(m), int(s))
            else:
                print '{0},{2:02d}:{3:02d}:{4:02d}'.format(jb, int(d), int(h), int(m), int(s))

然后运行 python ./util1.py f2.txt > f2-1.txt 这将输出保存到 f2-1.txt

然后

cp f1.txt f1_.txt

从f1_.txt删除标题行Job,Start,End

sort f1_.txt > f1.sorted.txt

sort f2-1.txt > f2-1.sorted.txt

和 diff -u f1.sorted.txt f2-1.sorted.txt

Answer 4

我正在尝试开发一些东西，您可以将其中一个文件拆分为较小的文件（例如每个文件 100,000 条记录），并保留每个文件的腌制字典，其中包含所有 Job_id 作为键，其行作为值.从某种意义上说，每个数据库都有一个索引，您可以对每个子文件使用哈希查找来确定是否要读取其内容。

但是，您说文件不断增长并且每个Job_id 都是唯一的。所以，我会咬紧牙关，运行你当前的分析一次。有一个行计数器，记录您为每个文件分析了多少行，并在某处写入文件。然后在将来，您可以使用linecache 来了解您想在file1 和file2 的下一次分析中从哪一行开始；之前的所有行都已处理，因此再次扫描该文件的全部内容绝对没有意义，只需从您在之前分析中结束的地方开始。

如果您以足够频繁的时间间隔运行分析，谁在乎它是否为 O(n^2)，因为您一次处理 10 条记录并将其附加到您的组合数据库中。换句话说，第一次分析需要很长时间，但随后的每个分析都会变得更快，最终n 应该会收敛到1，因此它变得无关紧要。

Answer 5

解析每个文件并将数据转换为datetime.timedelta 对象。制作一个以作业号为键、timedelta 对象为值的字典：

import operator, datetime, collections
def parse1(fp = 'job-file1.txt'):
    with open(fp) as f:
        next(f)
        for line in f:
            line = line.strip()
            job, job_length = line.split(',',1)
            if '-' in job_length:
                days, time = job_length.split('-')
                hours, minutes, seconds = time.split(':')
            else:
                days = 0
                hours, minutes, seconds = job_length.split(':')
            job_length = datetime.timedelta(days = int(days),
                                            hours = int(hours),
                                            minutes = int(minutes),
                                            seconds = int(seconds))
            yield (job, job_length)

fmt = '%Y-%m-%dT%H:%M:%S'
def parse2(fp = 'job-file2.txt'):
    with open(fp) as f:
        next(f)
        for line in f:
            line = line.strip()
            job, start, end = line.split(',')
            job_length = datetime.datetime.strptime(end, fmt) - datetime.datetime.strptime(start, fmt)
            yield (job, job_length)

现在您可以保留两个 timedelta 对象并稍后进行比较：

# {job_number:[timedelta1, timedelta2]
d = collections.defaultdict(list)

for key, value in parse1():
    d[key].append(value)

for key, value in parse2():
    d[key].append(value)

这让您可以执行以下操作：

differences = {job:lengths for job,lengths in d.items() if not operator.eq(*lengths)}
print(differences)

或者您可以只保留 file1 和 file2 作业长度之间的差异作为值

d = {key:value for key, value in parse1()}
for key, value in parse2():
    d[key] -= value

然后您只需检查与

的差异

[job for job, difference in d.items() if difference.total_seconds() != 0]