Python 3 连接来自已排序的大文件的数据

Question

我有多个按唯一时间戳排序的大文件（> 5M 行数据）。除了少数随机丢失的行（

除了缺少的行，我可以使用 zip：

def get_data(list_of_iterables):
    for data in zip(*list_of_iterables):
        yield data

但是，由于缺少一些行，我需要在时间戳上加入数据，而不是简单地压缩。我可以简单地忽略每个文件中没有匹配时间戳的任何行。

有没有一种pythonic 方法可以用几行代码来实现这个功能？

我的方法是依次推进每个可迭代对象，直到其时间戳不再小于可迭代对象组的最大时间戳。每当所有时间戳匹配时，产生一行并推进所有迭代。但是，当我尝试实现这种方法时，逻辑似乎很混乱。

编辑：性能。

实现需要在不先将所有数据读入内存的情况下开始返回行。读取所有数据需要一段时间，而且很多时候只需要检查前几行。

Answer 1

我最终编写了以下代码来解决我的问题，结果证明它比我预期的要轻：

def advance_values(iters):
    for it in iters:
        yield next(it)

def align_values(iters, values, key):
    for it, value in zip(iters, values):
        while (value[0],value[1]) < key:
            value = next(it)
        yield value

def merge_join(*iters):
    values = list(advance_values(iters))
    while True:
        if len(values) != len(iters):
            return
        tms = [(v[0],v[1]) for v in values]
        max_tm = max(tms)
        if all((v[0],v[1]) == max_tm for v in values):
            yield values
            values = list(advance_values(iters))
        else:
            values = list(align_values(iters, values, max_tm))

Answer 2

如果list_of_iterables 中的每个可迭代对象都按timestamp 排序，那么您可以使用heapq.merge() 合并它们，同时考虑到数据中可能存在的间隙，并使用itertools.groupby() 对具有相同时间戳的记录进行分组：

from heapq import merge
from itertools import groupby
from operator import attrgetter

for timestamp, group in groupby(merge(*list_of_iterables), 
                                key=attrgetter('timestamp')):
    print(timestamp, list(group)) # same timestamp

该实现无需先将所有数据读入内存即可生成组。

Answer 3

我的第一个猜测是使用带有时间戳的字典作为键，行中的其余数据作为值，然后对于每个文件中的每一行，仅当具有相同时间戳的项目（ key) 不存在。

但是，如果您真的在处理巨大的数据集（在这种情况下似乎就是这样），那么您在原始问题中提到的方法将是您的最佳选择。

Answer 4

好的，我对这个问题很感兴趣（最近遇到了一个类似的问题）并做了一些工作。你可以试试这样的：

import io
import datetime
from csv import DictReader

file0 = io.StringIO('''timestamp,data
2015-06-01 10:00, data00
2015-06-01 11:00, data01
2015-06-01 12:00, data02
2015-06-01 12:30, data03
2015-06-01 13:00, data04
''')

file1 = io.StringIO('''timestamp,data
2015-06-01 09:00, data10
2015-06-01 10:30, data11
2015-06-01 11:00, data12
2015-06-01 12:30, data13
''')

class Data(object):

    def __init__(self):
        self.timestamp = None
        self.data = None

    @staticmethod
    def new_from_dict(dct=None):
        if dct is None:
            return None
        ret = Data()
        ret.data = dct['data'].strip()
        ret.timestamp = datetime.datetime.strptime(dct['timestamp'],
                                                   '%Y-%m-%d %H:%M')
        return ret

    def __lt__(self, other):
        if other is None:
            return False
        return self.timestamp < other.timestamp

    def __gt__(self, other):
        if other is None:
            return False
        return self.timestamp > other.timestamp

    def __str__(self):
        ret = '{0.__class__.__name__}'.format(self) +\
              '(timestamp={0.timestamp}, data={0.data})'.format(self)
        return ret


def next_or_none(reader):
    try:
        return Data.new_from_dict(next(reader))
    except StopIteration:
        return None


def yield_in_order(reader0, reader1):

    data0 = next_or_none(reader0)
    data1 = next_or_none(reader1)

    while not data0 == data1 == None:

        if data0 is None:
            yield None, data1
            data1 = next_or_none(reader1)
            continue
        if data1 is None:
            yield data0, None
            data0 = next_or_none(reader0)
            continue

        while data0 < data1:
            yield data0, None
            data0 = next_or_none(reader0)

        while data0 > data1:
            yield None, data1
            data1 = next_or_none(reader1)

        if data0 is not None and data1 is not None:
            if data0.timestamp == data1.timestamp:
                yield data0, data1
                data0 = next_or_none(reader0)
                data1 = next_or_none(reader1)

csv0 = DictReader(file0)
csv1 = DictReader(file1)

FMT = '{!s:50s} | {!s:50s}'
print(FMT.format('file0', 'file1'))
print(101*'-')
for dta0, dta1 in yield_in_order(csv0, csv1):
    print(FMT.format(dta0, dta1))

这仅适用于 2 个文件。