PySpark批量解析格式混乱CSV：提取表头与数据

场景说明 #

原始CSV存在以下问题：

• 数据行被拆分到多个_c前缀的列中
• 每个列的单元格内容为「键:值」格式（如MANDT:400）
• 需要完成：批量重命名表头为键名、提取值填充对应列

解决方案1：同一列键统一场景 #

适用于每个_c列的所有单元格键名一致的情况（如_c0列全为MANDT:xxx）

import pandas as pd

# 读取原始CSV
df = pd.read_csv("messy_data.csv")

# 筛选所有_c前缀的目标列
target_cols = [col for col in df.columns if col.startswith("_c")]

result_data = {}

for col in target_cols:
    # 按第一个冒号拆分键值，避免值中含冒号导致的错误
    split_df = df[col].str.split(":", n=1, expand=True)
    # 提取该列的键名（取第一行的键）
    key_name = split_df[0].iloc[0]
    # 提取对应的值列
    result_data[key_name] = split_df[1]

# 生成清洗后的DataFrame并保存
clean_df = pd.DataFrame(result_data)
clean_df.to_csv("clean_data.csv", index=False)

解决方案2：同一列键不固定场景 #

适用于_c列中不同行的键名可能不同的情况

import pandas as pd

df = pd.read_csv("messy_data.csv")
target_cols = [col for col in df.columns if col.startswith("_c")]

clean_df = pd.DataFrame()

# 逐行处理每个单元格的键值对
for idx, row in df.iterrows():
    row_dict = {}
    for col in target_cols:
        cell_val = row[col]
        # 跳过空值，避免拆分报错
        if pd.notna(cell_val):
            key, val = cell_val.split(":", 1)
            row_dict[key] = val
    # 将当前行数据合并到结果DataFrame
    clean_df = pd.concat([clean_df, pd.DataFrame([row_dict])], ignore_index=True)

clean_df.to_csv("clean_data.csv", index=False)

常见报错处理 #

如果之前触发TypeError，大概率是以下原因：

1. 存在空值单元格：代码中加入了pd.notna()判断跳过空值
2. 值中包含冒号：使用split(":", n=1)仅按第一个冒号拆分，保留值中的冒号

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