PySpark实现单文件多数据集拆分：分存表头、明细、尾部DataFrame

核心思路 #

先按行读取文件，识别每组的表头（H行）、明细（D行）、尾部（T行），为每组分配唯一关联Key，再分别解析各类行的固定宽度字段，最后将三类数据转换为带共同Key的DataFrame。

具体实现步骤（Python + Pandas） #

1. 定义字段拆分规则 #

根据你的示例，明确各类行的固定宽度拆分逻辑：

• H行：第1位为标识（H），第2-5位是Attribute1，第6-12位是Attribute2，第13位及以后是Attribute3
• D行：第1位为标识（D），第2-4位是Attribute1，剩余部分为其他明细字段
• T行：第1位为标识（T），第2-5位是Attribute1，第6位及以后是Count

2. 读取并分组处理文件内容 #

遍历文件每一行，维护当前组的Key和各类记录：

import pandas as pd

# 初始化存储容器
header_records = []
detail_records = []
trailer_records = []
current_group_key = 0

# 读取目标文件（替换为你的文件路径）
with open('fixed_width_file.txt', 'r') as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        
        line_type = line[0]
        if line_type == 'H':
            # 开启新分组，更新Key
            current_group_key += 1
            # 解析H行字段
            attr1 = line[1:5]
            attr2 = line[5:12]
            attr3 = line[12:]
            header_records.append({
                'Key': current_group_key,
                'Attribute1': attr1,
                'Attribute2': attr2,
                'Attribute3': attr3
            })
        elif line_type == 'D':
            # 解析D行字段
            attr1 = line[1:4]
            other_attrs = line[4:]
            detail_records.append({
                'Key': current_group_key,
                'Attribute1': attr1,
                'OtherAttributes': other_attrs
            })
        elif line_type == 'T':
            # 解析T行字段
            attr1 = line[1:5]
            count = line[5:]
            trailer_records.append({
                'Key': current_group_key,
                'Attribute1': attr1,
                'Count': count
            })

3. 转换为目标DataFrame #

将收集到的记录列表转换为对应DataFrame：

# 表头DataFrame
header_df = pd.DataFrame(header_records)
# 明细DataFrame
detail_df = pd.DataFrame(detail_records)
# 尾部DataFrame
trailer_df = pd.DataFrame(trailer_records)

4. 验证关联有效性 #

可以通过Key验证三组数据的关联逻辑是否正确，比如检查每组明细数量是否与尾部记录的Count匹配：

# 统计每组实际明细数量
actual_detail_count = detail_df.groupby('Key').size().reset_index(name='ActualCount')
# 与尾部记录的Count做对比
validation_result = pd.merge(trailer_df, actual_detail_count, on='Key')
# 输出不匹配的组（如果有）
print(validation_result[validation_result['Count'] != validation_result['ActualCount'].astype(str)])

注意事项 #

• 若实际固定宽度规则与示例不同，只需调整字段拆分的索引位置即可
• 处理超大文件时，可考虑分批次写入数据，避免内存占用过高
• 若文件存在空行或格式异常行，建议增加异常处理逻辑（比如跳过不符合格式的行）

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