如何用Python处理大文件？

for

def process_large_text_file_line_by_line(filepath):     processed_count = 0     with open(filepath, 'r', encoding='utf-8') as f:         for line in f:             # 在这里对每一行进行处理             # 比如清洗数据、提取信息、写入新文件等             # print(f"Processing line: {line.strip()}")             processed_count += 1             if processed_count % 100000 == 0:                 print(f"Processed {processed_count} lines...")     print(f"Finished processing {processed_count} lines.")  # 示例调用 # process_large_text_file_line_by_line('your_large_file.txt')

def process_large_binary_file_in_chunks(filepath, chunk_size=4096): # 4KB chunks     processed_bytes = 0     with open(filepath, 'rb') as f:         while True:             chunk = f.read(chunk_size)             if not chunk:                 break # 文件读取完毕             # 在这里处理读取到的二进制块             # print(f"Processing chunk of size: {len(chunk)} bytes")             processed_bytes += len(chunk)             # 假设你可能需要将这些块写入另一个文件             # output_file.write(chunk)     print(f"Finished processing {processed_bytes} bytes.")  # 示例调用 # process_large_binary_file_in_chunks('your_large_binary_file.bin')

file.read()

file.readlines()

MemoryError

for line in file_object:

import csv  def filter_large_csv(filepath, condition_func):     """     一个生成器函数，用于从大型CSV文件中筛选行。     condition_func 是一个接受一行数据（字典）并返回True/False的函数。     """     with open(filepath, 'r', encoding='utf-8') as f:         reader = csv.DictReader(f) # 使用DictReader方便按列名访问         for row in reader:             if condition_func(row):                 yield row # 满足条件的行才会被生成  # 假设我们有一个名为 'large_data.csv' 的文件 # 并且我们想筛选 'age' 列大于30的行 def is_older_than_30(row):     try:         return int(row.get('age', 0)) > 30     except ValueError:         return False # 处理非数字的age值  # 实际使用时： # for filtered_row in filter_large_csv('large_data.csv', is_older_than_30): #     print(filtered_row) #     # 在这里对筛选出的行进行进一步处理 #     # 比如写入新的CSV文件，或者进行统计分析

filter_large_csv

pandas

pandas.read_csv()

import pandas as pd  def process_large_csv_with_pandas_chunks(filepath, chunk_size=10000):     """     使用pandas分块读取并处理大型CSV文件。     """     total_rows_processed = 0     # iterator=True 返回一个TextFileReader对象，可以迭代     for chunk_df in pd.read_csv(filepath, chunksize=chunk_size):         # chunk_df 是一个DataFrame，包含chunk_size行数据         # 在这里对每个数据块进行处理         # 例如，计算每块的平均值，筛选数据，或者写入数据库         print(f"Processing a chunk of {len(chunk_df)} rows. Current total: {total_rows_processed + len(chunk_df)}")         # 示例：计算某个列的平均值         # if 'value_column' in chunk_df.columns:         #     avg_value = chunk_df['value_column'].mean()         #     print(f"Average value in this chunk: {avg_value}")          total_rows_processed += len(chunk_df)     print(f"Finished processing {total_rows_processed} rows in total.")  # 示例调用 # process_large_csv_with_pandas_chunks('your_large_data.csv')

chunk_df

import re  def parse_large_log_file(filepath, pattern):     """     逐行读取大型日志文件，并使用正则表达式提取匹配的内容。     """     compiled_pattern = re.compile(pattern)     with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:         for line_num, line in enumerate(f, 1):             match = compiled_pattern.search(line)             if match:                 # 假设你只想提取匹配到的所有捕获组                 yield (line_num, match.groups())             # 也可以在这里处理没有匹配到的行，或者根据业务逻辑进行其他操作  # 示例：提取apache日志中的IP地址和请求路径 # log_pattern = r'^(d{1,3}.d{1,3}.d{1,3}.d{1,3}) - - [.*?] "GET (.+?) HTTP' # for line_num, data in parse_large_log_file('access.log', log_pattern): #     ip_address, request_path = data #     # print(f"Line {line_num}: IP={ip_address}, Path={request_path}")

import mmap import os  def search_in_large_file_with_mmap(filepath, search_term):     """     使用mmap在大型文件中查找特定字符串。     """     with open(filepath, 'r+b') as f: # 'r+b' 允许读写二进制模式         # mmap.mmap(fileno, length, access=ACCESS_READ)         # length=0 意味着映射整个文件         with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:             # 搜索是字节串，所以search_term需要编码             index = mm.find(search_term.encode('utf-8'))             if index != -1:                 print(f"Found '{search_term}' at byte offset: {index}")                 # 你可以进一步读取该位置附近的数据                 # 例如：print(mm[index:index+50].decode('utf-8'))             else:                 print(f"'{search_term}' not found.")  # 示例调用 # if os.path.exists('your_large_text_file.txt'): #     search_in_large_file_with_mmap('your_large_text_file.txt', 'important_keyword')