处理大文件的最快方法？
Fastest way to process a large file?
我有多个 3 GB 的制表符分隔文件。每个文件中有 2000 万行。所有行都必须独立处理，任何两行之间没有关系。我的问题是，什么会更快？
逐行读取？
with open() as infile:
for line in infile:
以块的形式将文件读入内存并处理它，比如一次 250 MB？
处理不是很复杂，我只是在column1到List1，column2到List2等中抓取值。可能需要将一些列值加在一起。
我在具有 30GB 内存的 linux 盒子上使用 python 2.7。 ASCII 文本。
有什么方法可以并行加速吗？现在我用的是前一种方法，过程很慢。使用任何 CSVReader 模块会有帮助吗？
我不必在 python 中这样做，欢迎使用任何其他语言或数据库的想法。
听起来您的代码已 I/O 绑定。这意味着多处理不会有帮助 — 如果您花费 90% 的时间从磁盘读取，那么让额外的 7 个进程等待下一次读取不会有任何帮助。
而且，虽然使用 CSV 读取模块（无论是标准库的 csv 还是 NumPy 或 Pandas 之类的东西）可能是一个简单的好主意，但它不太可能在性能上产生太大差异。
不过，值得检查一下您是否确实 I/O 绑定，而不仅仅是猜测。 运行你的程序，看看你的CPU使用率是接近0%还是接近100%还是一个核心。执行 Amadan 在评论中建议的操作，然后 运行 你的程序只需要 pass 进行处理，看看这会减少 5% 还是 70% 的时间。您甚至可能想尝试与 os.open 和 os.read(1024*1024) 上的循环进行比较，看看是否更快。
由于您使用 Python 2.x，Python 依赖于 C stdio 库来猜测一次要缓冲多少，因此可能值得强制它缓冲更多的。最简单的方法是对一些大的 bufsize 使用 readlines(bufsize)。 （您可以尝试不同的数字并测量它们以查看峰值在哪里。根据我的经验，通常 64K-8MB 之间的任何东西都差不多，但取决于您的系统可能会有所不同 - 特别是如果您正在阅读关闭具有高吞吐量但可怕延迟的网络文件系统，它淹没了实际物理驱动器的吞吐量与延迟以及OS所做的缓存。）
因此，例如：
bufsize = 65536
with open(path) as infile:
while True:
lines = infile.readlines(bufsize)
if not lines:
break
for line in lines:
process(line)
同时，假设您使用的是 64 位系统，您可能想尝试使用 mmap 而不是首先读取文件。这当然保证会更好，但可能会更好，具体取决于您的系统。例如：
with open(path) as infile:
m = mmap.mmap(infile, 0, access=mmap.ACCESS_READ)
A Python mmap 有点奇怪——它既像 str 又像 file，所以你可以，例如，手动迭代扫描换行符，或者您可以像调用文件一样对其调用 readline。与将文件作为行迭代或执行批处理 readlines 相比，这两者都需要从 Python 进行更多处理（因为 C 中的循环现在是纯 Python… 虽然也许你可以使用 re 或使用简单的 Cython 扩展来解决这个问题？）...但是 OS 的 I/O 优势知道您正在使用映射做什么可能会淹没 CPU 劣势。
不幸的是，Python 不会公开您用来调整内容以尝试在 C 中优化它的 madvise 调用（例如，显式设置 MADV_SEQUENTIAL让内核猜测，或强制透明大页面）——但实际上你可以 ctypes 函数出自 libc.
我知道这个问题很老了；但我想做类似的事情，我创建了一个简单的框架来帮助您并行读取和处理大文件。留下我尝试过的答案。
这是代码，我在最后举个例子
def chunkify_file(fname, size=1024*1024*1000, skiplines=-1):
"""
function to divide a large text file into chunks each having size ~= size so that the chunks are line aligned
Params :
fname : path to the file to be chunked
size : size of each chink is ~> this
skiplines : number of lines in the begining to skip, -1 means don't skip any lines
Returns :
start and end position of chunks in Bytes
"""
chunks = []
fileEnd = os.path.getsize(fname)
with open(fname, "rb") as f:
if(skiplines > 0):
for i in range(skiplines):
f.readline()
chunkEnd = f.tell()
count = 0
while True:
chunkStart = chunkEnd
f.seek(f.tell() + size, os.SEEK_SET)
f.readline() # make this chunk line aligned
chunkEnd = f.tell()
chunks.append((chunkStart, chunkEnd - chunkStart, fname))
count+=1
if chunkEnd > fileEnd:
break
return chunks
def parallel_apply_line_by_line_chunk(chunk_data):
"""
function to apply a function to each line in a chunk
Params :
chunk_data : the data for this chunk
Returns :
list of the non-None results for this chunk
"""
chunk_start, chunk_size, file_path, func_apply = chunk_data[:4]
func_args = chunk_data[4:]
t1 = time.time()
chunk_res = []
with open(file_path, "rb") as f:
f.seek(chunk_start)
cont = f.read(chunk_size).decode(encoding='utf-8')
lines = cont.splitlines()
for i,line in enumerate(lines):
ret = func_apply(line, *func_args)
if(ret != None):
chunk_res.append(ret)
return chunk_res
def parallel_apply_line_by_line(input_file_path, chunk_size_factor, num_procs, skiplines, func_apply, func_args, fout=None):
"""
function to apply a supplied function line by line in parallel
Params :
input_file_path : path to input file
chunk_size_factor : size of 1 chunk in MB
num_procs : number of parallel processes to spawn, max used is num of available cores - 1
skiplines : number of top lines to skip while processing
func_apply : a function which expects a line and outputs None for lines we don't want processed
func_args : arguments to function func_apply
fout : do we want to output the processed lines to a file
Returns :
list of the non-None results obtained be processing each line
"""
num_parallel = min(num_procs, psutil.cpu_count()) - 1
jobs = chunkify_file(input_file_path, 1024 * 1024 * chunk_size_factor, skiplines)
jobs = [list(x) + [func_apply] + func_args for x in jobs]
print("Starting the parallel pool for {} jobs ".format(len(jobs)))
lines_counter = 0
pool = mp.Pool(num_parallel, maxtasksperchild=1000) # maxtaskperchild - if not supplied some weird happend and memory blows as the processes keep on lingering
outputs = []
for i in range(0, len(jobs), num_parallel):
print("Chunk start = ", i)
t1 = time.time()
chunk_outputs = pool.map(parallel_apply_line_by_line_chunk, jobs[i : i + num_parallel])
for i, subl in enumerate(chunk_outputs):
for x in subl:
if(fout != None):
print(x, file=fout)
else:
outputs.append(x)
lines_counter += 1
del(chunk_outputs)
gc.collect()
print("All Done in time ", time.time() - t1)
print("Total lines we have = {}".format(lines_counter))
pool.close()
pool.terminate()
return outputs
比如说，我有一个文件，我想在其中计算每行中的单词数，那么每行的处理看起来像
def count_words_line(line):
return len(line.strip().split())
然后像这样调用函数：
parallel_apply_line_by_line(input_file_path, 100, 8, 0, count_words_line, [], fout=None)
与普通的逐行读取大小约 20GB 的示例文件相比，我使用它的速度提高了约 8 倍，我在其中对每一行进行了一些适度复杂的处理。
我有多个 3 GB 的制表符分隔文件。每个文件中有 2000 万行。所有行都必须独立处理，任何两行之间没有关系。我的问题是，什么会更快？
逐行读取？
with open() as infile: for line in infile:以块的形式将文件读入内存并处理它，比如一次 250 MB？
处理不是很复杂，我只是在column1到List1，column2到List2等中抓取值。可能需要将一些列值加在一起。
我在具有 30GB 内存的 linux 盒子上使用 python 2.7。 ASCII 文本。
有什么方法可以并行加速吗？现在我用的是前一种方法，过程很慢。使用任何 CSVReader 模块会有帮助吗？
我不必在 python 中这样做，欢迎使用任何其他语言或数据库的想法。
听起来您的代码已 I/O 绑定。这意味着多处理不会有帮助 — 如果您花费 90% 的时间从磁盘读取，那么让额外的 7 个进程等待下一次读取不会有任何帮助。
而且，虽然使用 CSV 读取模块（无论是标准库的 csv 还是 NumPy 或 Pandas 之类的东西）可能是一个简单的好主意，但它不太可能在性能上产生太大差异。
不过，值得检查一下您是否确实 I/O 绑定，而不仅仅是猜测。 运行你的程序，看看你的CPU使用率是接近0%还是接近100%还是一个核心。执行 Amadan 在评论中建议的操作，然后 运行 你的程序只需要 pass 进行处理，看看这会减少 5% 还是 70% 的时间。您甚至可能想尝试与 os.open 和 os.read(1024*1024) 上的循环进行比较，看看是否更快。
由于您使用 Python 2.x，Python 依赖于 C stdio 库来猜测一次要缓冲多少，因此可能值得强制它缓冲更多的。最简单的方法是对一些大的 bufsize 使用 readlines(bufsize)。 （您可以尝试不同的数字并测量它们以查看峰值在哪里。根据我的经验，通常 64K-8MB 之间的任何东西都差不多，但取决于您的系统可能会有所不同 - 特别是如果您正在阅读关闭具有高吞吐量但可怕延迟的网络文件系统，它淹没了实际物理驱动器的吞吐量与延迟以及OS所做的缓存。）
因此，例如：
bufsize = 65536
with open(path) as infile:
while True:
lines = infile.readlines(bufsize)
if not lines:
break
for line in lines:
process(line)
同时，假设您使用的是 64 位系统，您可能想尝试使用 mmap 而不是首先读取文件。这当然保证会更好，但可能会更好，具体取决于您的系统。例如：
with open(path) as infile:
m = mmap.mmap(infile, 0, access=mmap.ACCESS_READ)
A Python mmap 有点奇怪——它既像 str 又像 file，所以你可以，例如，手动迭代扫描换行符，或者您可以像调用文件一样对其调用 readline。与将文件作为行迭代或执行批处理 readlines 相比，这两者都需要从 Python 进行更多处理（因为 C 中的循环现在是纯 Python… 虽然也许你可以使用 re 或使用简单的 Cython 扩展来解决这个问题？）...但是 OS 的 I/O 优势知道您正在使用映射做什么可能会淹没 CPU 劣势。
不幸的是，Python 不会公开您用来调整内容以尝试在 C 中优化它的 madvise 调用（例如，显式设置 MADV_SEQUENTIAL让内核猜测，或强制透明大页面）——但实际上你可以 ctypes 函数出自 libc.
我知道这个问题很老了；但我想做类似的事情，我创建了一个简单的框架来帮助您并行读取和处理大文件。留下我尝试过的答案。
这是代码，我在最后举个例子
def chunkify_file(fname, size=1024*1024*1000, skiplines=-1):
"""
function to divide a large text file into chunks each having size ~= size so that the chunks are line aligned
Params :
fname : path to the file to be chunked
size : size of each chink is ~> this
skiplines : number of lines in the begining to skip, -1 means don't skip any lines
Returns :
start and end position of chunks in Bytes
"""
chunks = []
fileEnd = os.path.getsize(fname)
with open(fname, "rb") as f:
if(skiplines > 0):
for i in range(skiplines):
f.readline()
chunkEnd = f.tell()
count = 0
while True:
chunkStart = chunkEnd
f.seek(f.tell() + size, os.SEEK_SET)
f.readline() # make this chunk line aligned
chunkEnd = f.tell()
chunks.append((chunkStart, chunkEnd - chunkStart, fname))
count+=1
if chunkEnd > fileEnd:
break
return chunks
def parallel_apply_line_by_line_chunk(chunk_data):
"""
function to apply a function to each line in a chunk
Params :
chunk_data : the data for this chunk
Returns :
list of the non-None results for this chunk
"""
chunk_start, chunk_size, file_path, func_apply = chunk_data[:4]
func_args = chunk_data[4:]
t1 = time.time()
chunk_res = []
with open(file_path, "rb") as f:
f.seek(chunk_start)
cont = f.read(chunk_size).decode(encoding='utf-8')
lines = cont.splitlines()
for i,line in enumerate(lines):
ret = func_apply(line, *func_args)
if(ret != None):
chunk_res.append(ret)
return chunk_res
def parallel_apply_line_by_line(input_file_path, chunk_size_factor, num_procs, skiplines, func_apply, func_args, fout=None):
"""
function to apply a supplied function line by line in parallel
Params :
input_file_path : path to input file
chunk_size_factor : size of 1 chunk in MB
num_procs : number of parallel processes to spawn, max used is num of available cores - 1
skiplines : number of top lines to skip while processing
func_apply : a function which expects a line and outputs None for lines we don't want processed
func_args : arguments to function func_apply
fout : do we want to output the processed lines to a file
Returns :
list of the non-None results obtained be processing each line
"""
num_parallel = min(num_procs, psutil.cpu_count()) - 1
jobs = chunkify_file(input_file_path, 1024 * 1024 * chunk_size_factor, skiplines)
jobs = [list(x) + [func_apply] + func_args for x in jobs]
print("Starting the parallel pool for {} jobs ".format(len(jobs)))
lines_counter = 0
pool = mp.Pool(num_parallel, maxtasksperchild=1000) # maxtaskperchild - if not supplied some weird happend and memory blows as the processes keep on lingering
outputs = []
for i in range(0, len(jobs), num_parallel):
print("Chunk start = ", i)
t1 = time.time()
chunk_outputs = pool.map(parallel_apply_line_by_line_chunk, jobs[i : i + num_parallel])
for i, subl in enumerate(chunk_outputs):
for x in subl:
if(fout != None):
print(x, file=fout)
else:
outputs.append(x)
lines_counter += 1
del(chunk_outputs)
gc.collect()
print("All Done in time ", time.time() - t1)
print("Total lines we have = {}".format(lines_counter))
pool.close()
pool.terminate()
return outputs
比如说，我有一个文件，我想在其中计算每行中的单词数，那么每行的处理看起来像
def count_words_line(line):
return len(line.strip().split())
然后像这样调用函数：
parallel_apply_line_by_line(input_file_path, 100, 8, 0, count_words_line, [], fout=None)
与普通的逐行读取大小约 20GB 的示例文件相比，我使用它的速度提高了约 8 倍，我在其中对每一行进行了一些适度复杂的处理。