我想逐行读取一个大文件(>5GB),而不需要将其整个内容加载到内存中。由于readlines()
会在内存中创建一个非常大的列表,因此我无法使用它。
使用 for
循环读取文件对象的每一行。使用 with open(...)
来让一个上下文管理器确保在读取后关闭文件:
with open("log.txt") as infile:
for line in infile:
print(line)
chunk = infile.read(chunksize)
读取有限大小的块,而不管它们的内容,而不是使用迭代行的for
循环。您将不得不自己在块内搜索换行符。 - John La Rooy您所需要做的就是将文件对象用作迭代器。
for line in open("log.txt"):
do_something_with(line)
在最近的Python版本中,使用上下文管理器会更好。
with open("log.txt") as fileobject:
for line in fileobject:
do_something_with(line)
这也会自动关闭文件。
with
语句是一个“上下文管理器”,它会自动关闭文件对象。 - Keithwith open('filename','r',buffering=100000) as f:
for line in f:
print line
最好使用迭代器。
相关文档:fileinput
— 从多个输入流中迭代读取行。
来自文档:
import fileinput
for line in fileinput.input("filename", encoding="utf-8"):
process(line)
这样做将避免一次性将整个文件复制到内存中。
FileInput
类对象的close()
方法,因此我建议避免以这种方式使用它。在Python 3.2中,他们终于使fileinput
与上下文管理器协议兼容,解决了这个问题(但代码仍不能完全按照所示方式编写)。 - martineau一种老派的方法:
fh = open(file_name, 'rt')
line = fh.readline()
while line:
# do stuff with line
line = fh.readline()
fh.close()
如果文件中没有换行符,以下是您需要执行的操作:
with open('large_text.txt') as f:
while True:
c = f.read(1024)
if not c:
break
print(c,end='')
我简直不敢相信这件事情可以像@john-la-rooy的回答所说的那样容易。因此,我使用逐行读写的方式重新创建了cp
命令。它非常快速。
#!/usr/bin/env python3.6
import sys
with open(sys.argv[2], 'w') as outfile:
with open(sys.argv[1]) as infile:
for line in infile:
outfile.write(line)
readline
标准化了行结束符,这会将具有DOS行结束符\r\n
的文档转换为Unix行结束符\n
。我搜索这个主题的原因是我需要转换一个日志文件,该文件接收了各种.NET库产生的混乱行结束符。令我惊讶的是,在我的初始速度测试之后,我不需要返回并使用rstrip
来去除行末空白(\r\n
)。它已经完美了! - Bruno Bronoskyblaze项目在过去的6年中已经取得了很大的进展。它有一个简单的API,涵盖了pandas功能的一个有用子集。
dask.dataframe在内部处理分块,支持许多可并行化的操作,并允许您轻松地将切片导出回pandas以进行内存操作。
import dask.dataframe as dd
df = dd.read_csv('filename.csv')
df.head(10) # return first 10 rows
df.tail(10) # return last 10 rows
# iterate rows
for idx, row in df.iterrows():
...
# group by my_field and return mean
df.groupby(df.my_field).value.mean().compute()
# slice by column
df[df.my_field=='XYZ'].compute()
https://gist.github.com/iyvinjose/e6c1cb2821abd5f01fd1b9065cbc759d
下载文件data_loading_utils.py,并将其导入您的代码中。import data_loading_utils.py.py
file_name = 'file_name.ext'
CHUNK_SIZE = 1000000
def process_lines(data, eof, file_name):
# check if end of file reached
if not eof:
# process data, data is one single line of the file
else:
# end of file reached
data_loading_utils.read_lines_from_file_as_data_chunks(file_name, chunk_size=CHUNK_SIZE, callback=self.process_lines)
def readJSON_line2(linesIn):
#Function for reading a chunk of json lines
'''
Note, this function is nonsensical. A user would never use the approach suggested
for reading in a JSON file,
its role is to evaluate the MT approach for full line by line processing to both
increase speed and reduce memory overhead
'''
import json
linesRtn = []
for lineIn in linesIn:
if lineIn.strip() != 0:
lineRtn = json.loads(lineIn)
else:
lineRtn = ""
linesRtn.append(lineRtn)
return linesRtn
# -------------------------------------------------------------------
if __name__ == "__main__":
import multiprocessing as mp
path1 = "C:\\user\\Documents\\"
file1 = "someBigJson.json"
nBuffer = 20*nCPUs # How many chunks are queued up (so cpus aren't waiting on processes spawning)
nChunk = 1000 # How many lines are in each chunk
#Both of the above will require balancing speed against memory overhead
iJob = 0 #Tracker for SMP jobs submitted into pool
iiJob = 0 #Tracker for SMP jobs extracted back out of pool
jobs = [] #SMP job holder
MTres3 = [] #Final result holder
chunk = []
iBuffer = 0 # Buffer line count
with open(path1+file1) as f:
for line in f:
#Send to the chunk
if len(chunk) < nChunk:
chunk.append(line)
else:
#Chunk full
#Don't forget to add the current line to chunk
chunk.append(line)
#Then add the chunk to the buffer (submit to SMP pool)
jobs.append(pool.apply_async(readJSON_line2, args=(chunk,)))
iJob +=1
iBuffer +=1
#Clear the chunk for the next batch of entries
chunk = []
#Buffer is full, any more chunks submitted would cause undue memory overhead
#(Partially) empty the buffer
if iBuffer >= nBuffer:
temp1 = jobs[iiJob].get()
for rtnLine1 in temp1:
MTres3.append(rtnLine1)
iBuffer -=1
iiJob+=1
#Submit the last chunk if it exists (as it would not have been submitted to SMP buffer)
if chunk:
jobs.append(pool.apply_async(readJSON_line2, args=(chunk,)))
iJob +=1
iBuffer +=1
#And gather up the last of the buffer, including the final chunk
while iiJob < iJob:
temp1 = jobs[iiJob].get()
for rtnLine1 in temp1:
MTres3.append(rtnLine1)
iiJob+=1
#Cleanup
del chunk, jobs, temp1
pool.close()