如何编写PySpark脚本将邮件内容转为CSV适用的长字符串？

你的问题核心是正则未彻底清除所有换行/回车类控制字符，加上Spark写CSV的默认配置未做强化，导致保存后内容被拆分，下面是具体解决步骤：

1. 修正正则表达式，彻底清理冲突字符 #

现有正则没覆盖全所有会导致CSV拆分的控制字符，替换成以下代码，确保清除换行、回车、制表符、逗号、双引号，并合并连续短横线：

from pyspark.sql.functions import regexp_replace

# 先替换所有冲突字符为短横线，再合并连续的短横线
t = maintnence_request_df.withColumn(
    'Notes',
    regexp_replace(
        regexp_replace('Notes', r'[\n\r\t,"]+', '-'),
        r'-+', '-'
    )
)

• [\n\r\t,"]+ 匹配所有换行(\n)、回车(\r)、制表符(\t)、CSV分隔符(,)、双引号(")，这些是触发CSV行拆分的关键字符
• 第二层正则把连续的-合并为单个，避免冗余符号

2. 强化Spark写CSV的配置 #

即使替换了字符，添加以下配置确保字段被正确包裹，避免漏网字符触发拆分：
修改写CSV的代码段，增加quoteAll和escape选项：

df.coalesce(1).write.option("header", "true") \
    .option("quoteAll", "true")  # 给所有字段添加引号，强制CSV阅读器识别为单个字段
    .option("escape", "\"")  # 转义字段内可能残留的双引号
    .mode('overwrite').format(file_format).save(path)

3. 修复自定义保存函数的小错误 #

你的重命名代码里fs.delete(src_path, True)是多余的，rename操作已经移动了原文件，原路径不存在会报错，直接删除这一行即可。

测试时先验证处理后的DataFrame中Notes列是否为纯单行，再保存CSV即可解决拆分问题。

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