如何获取大文件的大致行数答案

【问题标题】：How to get approximate line number of large files如何获取大文件的大致行数
【发布时间】：2018-03-16 14:57:13
【问题描述】：

我有多达 10M+ 行的 CSV 文件。我正在尝试获取文件的总行号，以便可以将每个文件的处理拆分为多处理方法。为此，我将为每个要处理的子流程设置一个开始和结束行。对于 2GB 的文件大小，这将我的处理时间从 180 秒减少到 110 秒。但是，为了做到这一点，它需要知道行号数。如果我尝试获取确切的行号计数，则需要大约 30 秒。我觉得这段时间被浪费了，因为最终线程可能不得不读取额外的十万行左右，与获得确切行数所需的 30 秒相比，只会增加几秒钟。

如何获取文件的大致行数？我希望这个估计在 100 万行以内（最好在几十万行以内）。这样的事情可能吗？

【问题讨论】：

所有行的长度是否大致相同？
@khelwood 他们应该是，因为它都是表格数据。
然后将文件总长度除以任何一行的长度，即可得出大致的行数
您是否尝试过编写此代码？您可以从 pandas 库开始。
尝试一个 mapreduce 作业，它会自行将数据分成相等的部分。

标签： python filereader

【解决方案1】：

这将非常不准确，但它会得到一行的大小并将其除以文件的大小。

import sys
import csv
import os

with open("example.csv", newline="") as f:
    reader = csv.reader(f)
    row1   = next(reader)

    _Size = sys.getsizeof(len("".join(row1)))

print("Size of Line 1 > ",_Size)
print("Size of File   >",str(os.path.getsize("example.csv")))
print("Approx Lines   >",(os.path.getsize("example.csv") / _Size))

（编辑）如果将最后一行更改为 math.floor(os.path.getsize("example.csv") / _Size)其实是相当准确

【讨论】：

谢谢，我想我会使用这样的方法，但取前 100 行的平均值，这似乎是我能想到的最佳方法。

【解决方案2】：

我建议您在解析之前将文件分成大小相似的块。

下面的示例代码将通过查找和搜索下一个换行符将data.csv 分成大小大致相等的 4 块。然后它会为每个块调用launch_worker()，指明worker应该处理的数据的起始偏移量和长度。

理想情况下，您应该为每个工作人员使用 subprocess。

import os

n_workers = 4

# open the log file, and find out how long it is
f = open('data.csv', 'rb')
length_total = f.seek(0, os.SEEK_END)

# split the file evenly among n workers
length_worker = int(length_total / n_workers)

prev_worker_end = 0

for i in range(n_workers):
    # seek to the next worker's approximate start
    file_pos = f.seek(prev_worker_end + length_worker, os.SEEK_SET)

    # see if we tried to seek past the end of the file... the last worker probably will
    if file_pos >= length_total:                                            # <-- (3)
        # ... if so, this worker's chunk extends to the end of the file
        this_worker_end = length_total

    else:
        # ... otherwise, look for the next line break
        buf = f.read(256)                                                   # <-- (1)
        next_line_end = buf.index(b'\n')                                    # <-- (2)

        this_worker_end = file_pos + next_line_end

    # calculate how long this worker's chunk is
    this_worker_length = this_worker_end - prev_worker_end
    if this_worker_length > 0:
        # if there is any data in the chunk, then try to launch a worker
        launch_worker(prev_worker_end, this_worker_length)

    # remember where the last worker got to in the file
    prev_worker_end = this_worker_end + 1

代码中标记的一些扩展：

您需要确保read() 至少会占用一整行。或者，如果您不知道一行可以提前多长时间，您可以循环执行多个 read()s。
这假定\n 行结尾...您可能需要针对您的数据进行修改。
最后一个工作人员处理的数据比其他工作人员要少...这是因为我们总是向前搜索下一个换行符。你拥有的工人越多，最终工人获得的数据就越少。这不是很重要（在我的测试中约为 200-500 字节）。

确保您始终使用二进制模式，因为文本模式会给您带来不稳定的seek()s / read()s。

launch_worker() 的示例如下所示：

def launch_worker(offset, length):
    print('Starting a worker... using chunk %d - %d (%d bytes)...' 
           % ( offset, offset + length, length ))

    with open('log.txt', 'rb') as f:
        f.seek(offset, os.SEEK_SET)
        worker_buf = f.read(length)

    lines = worker_buf.split(b'\n')

    print('First Line:')
    print('\t' + str(lines[0]))
    print('Last Line:')
    print('\t' + str(lines[-1]))

【讨论】：