如何用Python或PySpark高效拆分大文本并按Schema处理记录

现有15GB大型文本文件，内容为单个连续字符串，包含约2000万条固定长度（5000字符）的记录，每条记录对应450+列。需要将每条记录拆分换行，并按指定Schema添加分隔符，最终转为DataFrame。当前实现因嵌套循环导致迭代量爆炸（2500万记录×500列），500MB文件处理耗时极久，多线程/进程无明显改善，寻求Python或PySpark低时间复杂度优化方案。

示例说明 #

• 输入示例：

HiIamRowData1HiIamRowData2HiIamRowData3HiIamRowData4HiIamRowData5HiIamRowData6HiIamRowData7HiIamRowData8

• 输出示例：

Hi#I#am#Row#Data#1#
Hi#I#am#Row#Data#2#
Hi#I#am#Row#Data#3#
Hi#I#am#Row#Data#4#
Hi#I#am#Row#Data#5#
Hi#I#am#Row#Data#6#
Hi#I#am#Row#Data#7#
Hi#I#am#Row#Data#8#

已尝试的Python代码 #

# Schema定义
schemaData = [['col1',0,2],['col2',2,1],['col3',3,2],['col4',5,3],['col5',8,4],['col6',12,1]]
df = pd.DataFrame(data= schemaData, columns=['FeildName','offset','size'])
print(df.head(5))

file = 'sampleText.txt'
inputFile = open(file, 'r').read()

recordLen = 13
totFileLen = len(inputFile)
finalStr = ''

# 按记录长度拆分每条记录
for i in range(0,totFileLen,recordLen):
    record = inputFile[i:i+recordLen]
    recStr = ''
    # 对每条记录应用Schema拆分列
    for index, row in df.iterrows():
        recStr = recStr + record[row['offset']:row['offset'] + row['size']] + '#'  
    recStr = recStr + '\n'
    finalStr += recStr
print(finalStr)

text_file = open("Output.txt", "w")
text_file.write(finalStr)

该代码嵌套循环导致迭代量剧增：示例中8条记录需56次迭代，实际场景为2500万×500次迭代，效率极低。

约束条件 #

• 文件为连续字符串，记录首尾相连，需完整读取并输出处理后的文本文件
• 不能按固定大小（如50MB）拆分文件块，避免记录被截断导致解析错误
• 仅可按记录长度拆分块处理

优化方案 #

一、Python端优化：减少循环开销+批量处理 #

核心思路是避免逐行逐列的嵌套循环，利用字符串切片的批量操作、预编译列规则，以及分块读写减少内存压力。

1. 预编译列提取规则

提前计算每条记录的列切片位置，避免循环中重复读取DataFrame：

# 预编译列提取规则：[(start, end), ...]
schemaData = [['col1',0,2],['col2',2,1],['col3',3,2],['col4',5,3],['col5',8,4],['col6',12,1]]
col_slices = [(row[1], row[1]+row[2]) for row in schemaData]
delimiter = '#'
record_len = 13

# 定义单条记录的处理函数（提前编译，避免循环内重复计算）
def process_record(record):
    parts = [record[s:e] for s,e in col_slices]
    return delimiter.join(parts) + delimiter + '\n'

2. 分块读取+批量处理

不一次性读取整个15GB文件，而是按记录长度×批量数分块读取，减少内存占用，同时用列表推导代替字符串拼接（字符串拼接为O(n²)操作，列表为O(n)）：

batch_size = 10000  # 每次处理1万条记录，可根据内存调整
block_size = record_len * batch_size

with open('sampleText.txt', 'r') as infile, open('Output.txt', 'w') as outfile:
    while True:
        block = infile.read(block_size)
        if not block:
            break
        # 拆分块内的记录
        records = [block[i:i+record_len] for i in range(0, len(block), record_len)]
        # 批量处理并写入
        processed_lines = [process_record(rec) for rec in records]
        outfile.writelines(processed_lines)

3. 用NumPy向量化操作进一步加速

如果内存允许，可将字符串转为NumPy数组，利用向量化操作处理：

import numpy as np

def process_block_vectorized(block, record_len, col_slices):
    # 将块转为字符数组，按记录拆分
    arr = np.array(list(block)).reshape(-1, record_len)
    # 提取每列的字符并拼接
    parts = []
    for s,e in col_slices:
        parts.append(''.join(arr[:, s:e].flatten()))
    # 按记录拆分拼接后的列
    processed = [delimiter.join([p[i*(e-s):(i+1)*(e-s)] for p,s,e in zip(parts, *zip(*col_slices))]) + delimiter + '\n' for i in range(arr.shape[0])]
    return processed

二、PySpark端优化：分布式并行处理 #

对于15GB级别的文件，PySpark的分布式处理是更优选择，天然避免单进程瓶颈。

1. 按记录长度拆分RDD

Spark默认按文件块拆分，需自定义输入格式保证记录不被截断：

from pyspark import SparkContext
from pyspark.sql import SparkSession

sc = SparkContext(appName="LargeTextParser")
spark = SparkSession(sc)

record_len = 5000  # 实际每条记录长度
delimiter = '#'

# 自定义输入格式：按记录长度读取
def read_records(file_path):
    with open(file_path, 'r') as f:
        while True:
            record = f.read(record_len)
            if not record:
                break
            yield record

# 创建RDD，切片数根据集群资源调整
records_rdd = sc.parallelize(read_records('large_file.txt'), numSlices=100)

# 预编译列规则
schemaData = [['col1',0,2],['col2',2,1],...]  # 实际Schema
col_slices = [(row[1], row[1]+row[2]) for row in schemaData]

# 处理每条记录
def process_spark_record(record):
    parts = [record[s:e] for s,e in col_slices]
    return delimiter.join(parts) + delimiter

processed_rdd = records_rdd.map(process_spark_record)

# 写入文件（Spark会自动分块写入）
processed_rdd.saveAsTextFile('output_dir')

# 转为DataFrame
from pyspark.sql.types import StringType, StructType, StructField
schema = StructType([StructField(row[0], StringType(), True) for row in schemaData])
# 先将处理后的字符串拆分回列
df = processed_rdd.map(lambda x: tuple(x.split(delimiter)[:-1])).toDF(schema=schema)
df.show()

2. 优化点说明

• 避免在Driver端读取整个文件，改用sc.parallelize结合本地读取生成RDD，或使用Spark的wholeTextFiles后再拆分记录（注意内存限制）
• 调整numSlices参数匹配集群CPU核心数，最大化并行度
• 如果需要直接生成DataFrame，可跳过字符串拼接步骤，直接在RDD中提取列元组，减少字符串处理开销

内容的提问来源于stack exchange，提问作者Anvesh