如何在PySpark中正确读取含多行内容的|分隔CSV文件

问题场景 #

我们有一个以|为分隔符的CSV文件，其中某条记录的SMTPBounceReason字段包含多行文本内容。使用PySpark默认的CSV读取器加载时，这些多行内容会被错误拆分成多条仅含单个字段值、其余字段为空的无效记录，导致DataFrame结构混乱。

原CSV数据示例 #

AccountID|BounceSubcategory|BounceTypeID|BounceType|SMTPBounceReason|SMTPMessage|SMTPCode|TriggererSendDefinitionObjectID|TriggeredSendCustomerKey|IsFalseBounce|dwHashKey
1111111111|Unknown|1|delayed|5.0.350  Remote server returned an error|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS107"}|580|||True|
1212121212|Mailbox Full|1|delayed|5.2.2  Generating server: MW4PR84MB2114.NAMPRD84.PROD.OUTLOOK.COM

peterspan@example.com
Remote server returned 554 5.2.2 mailbox full;
STOREDRV.Deliver.Exception:=
QuotaExceededException; Failed to process message due to a permanent
except=
ion wit|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS117"}|534|||False|

默认读取后的错误DataFrame #

|AccountID|BounceSubcategory|BounceTypeID|BounceType|SMTPBounceReason|SMTPMessage|SMTPCode|TriggererSendDefinitionObjectID|TriggeredSendCustomerKey|IsFalseBounce|dwHashKey
|1111111111|Unknown|1|delayed|5.0.350  Remote server returned an error|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS107"}|580|NULL|NULL|True|NULL|
|1212121212|Mailbox Full|1|delayed|5.2.2  Generating server: MW4PR84MB2114.NAMPRD84.PROD.OUTLOOK.COM|NULL|NULL|NULL|NULL|NULL|NULL|
|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|peterspan@example.com|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|Remote server returned 554 5.2.2 mailbox full;|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|STOREDRV.Deliver.Exception:=|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|QuotaExceededException; Failed to process message due to a permanent|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|except=|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|NULL|
|ion wit|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS117"}|534|NULL|NULL|False|NULL|NULL|NULL|NULL|

预期正确的DataFrame格式 #

AccountID|BounceSubcategory|BounceTypeID|BounceType|SMTPBounceReason|SMTPMessage|SMTPCode|TriggererSendDefinitionObjectID|TriggeredSendCustomerKey|IsFalseBounce|dwHashKey
1111111111|Unknown|1|delayed|5.0.350  Remote server returned an error|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS107"}|580|||True|
1212121212|Mailbox Full|1|delayed|5.2.2  Generating server: MW4PR84MB2114.NAMPRD84.PROD.OUTLOOK.COM peterspan@example.com Remote server returned 554 5.2.2 mailbox full; STOREDRV.Deliver.Exception:= QuotaExceededException; Failed to process message due to a permanent except=ion wit|{"Source":"MITS_SMTP", "Machine":"ATL1S11MITS117"}|534|||False| 

解决方案 #

核心思路：利用正常记录的字段数固定（11个）的特点，先按行读取所有文本，再通过字段数判断合并多行内容到对应记录的SMTPBounceReason字段中。

实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("MultilineCSVReader").getOrCreate()

# 定义CSV的Schema
schema = StructType([
    StructField("AccountID", StringType(), True),
    StructField("BounceSubcategory", StringType(), True),
    StructField("BounceTypeID", StringType(), True),
    StructField("BounceType", StringType(), True),
    StructField("SMTPBounceReason", StringType(), True),
    StructField("SMTPMessage", StringType(), True),
    StructField("SMTPCode", StringType(), True),
    StructField("TriggererSendDefinitionObjectID", StringType(), True),
    StructField("TriggeredSendCustomerKey", StringType(), True),
    StructField("IsFalseBounce", StringType(), True),
    StructField("dwHashKey", StringType(), True)
])

# 按文本行读取文件
lines = spark.read.text("path/to/your/file.csv").rdd.map(lambda x: x[0])

# 处理多行记录
processed_lines = []
current_record = None
target_field_count = len(schema.fields)

for line in lines.collect():
    # 跳过空行
    if not line.strip():
        continue
    # 分割当前行
    parts = line.split("|")
    # 判断是否为完整记录
    if len(parts) == target_field_count:
        # 如果有正在构建的记录，先加入结果
        if current_record is not None:
            processed_lines.append("|".join(current_record))
        # 初始化新的完整记录
        current_record = parts.copy()
    else:
        # 非完整记录，追加到上一条的SMTPBounceReason字段
        if current_record is not None:
            # 将当前行内容用空格连接到SMTPBounceReason末尾
            current_record[4] += " " + line.strip()

# 加入最后一条记录
if current_record is not None:
    processed_lines.append("|".join(current_record))

# 将处理后的行转换为DataFrame
df = spark.createDataFrame([line.split("|") for line in processed_lines], schema=schema)

# 查看结果
df.show(truncate=False)

代码说明 #

1. 读取文本行：先用spark.read.text按行读取所有内容，避免默认CSV读取器拆分多行字段。
2. 处理多行内容：遍历每行，通过分割后的字段数判断是否为完整记录。非完整记录直接追加到上一条记录的SMTPBounceReason字段（索引为4，对应Schema中的第5个字段）。
3. 转换为DataFrame：将处理后的完整记录列表转换为符合Schema的DataFrame。

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