对于这种情况,大文件,比如超过 10G,但内存有限,比如小于 2G,该如何处理,

标准读取流程如,

"""
计算文件行数
"""
count = 0
with open(fname) as datas:
  for data in datas:
    count += 1

其好处在于:

  • with 上下文管理器会自动关闭打开的文件描述符
  • 在迭代文件对象时,内容是一行一行返回的,不会占用太多内存

但,其缺点在于,如果被读取的文件里,没有任何换行符时,那么 data 将会变成一个非常巨大的字符串对象,内存消耗会非常大.

换一种做法,利用 chunk_size

count = 0
block_size = 1024 * 8 #每次 8kb
with open(fname) as fp:
  while True:
    chunk = fp.read(block_size)
    # 当文件没有更多内容时,read调用将会返回空字符串''
    if not chunk:
      break
    count += 1

可以发现,代码中存在着两个独立的逻辑:数据生成(read 调用与 chunk 判断)与数据消费,并将两个独立逻辑耦合在一起.

对此,可以采用如下生成器方式解耦合:

def chunked_file_reader(fp, block_size=1024 * 8):
  """
  生成器函数:分块读取文件内容
  """
  while True:
    chunk = fp.read(block_size)
    # 当文件没有更多内容时,read 调用将会返回空字符串''
    if not chunk:
      break
    yield chunk
#
count = 0
with open(fname) as fp:
  for chunk in chunked_file_reader(fp):
    count += 1

此外,python 内置函数 iter(iterable) 函数是用于构造迭代器的,在使用该函数进行调用时,会返回一个特殊对象,迭代它将不断产生可调用对应 callable 的调用结果,直到结果为setinel时,迭代终止,如:

def chunked_file_reader(file, block_size=1024 * 8):
  """
  生成器函数:分块读取文件内容,使用 iter 函数
  """
  #首先使用 partial(fp.read, block_size) 构造一个新的无需参数的函数
  #循环将不断返回 fp.read(block_size) 调用结果,直到其为''时终止
  for chunk in iter(partial(file.read, block_size), ''):
    yield chunk

只需要两行代码,就构造出了一个可复用的分块读取方法.

Last modification:April 11th, 2021 at 01:37 pm