如何在PySpark DataFrame中替换换行、退格及回车字符

正则匹配失败的核心原因是多数人会误用正则元字符\b匹配退格符，但\b在正则语法中默认代表单词边界，无法匹配字面意义上的ASCII退格控制符，必须用十六进制转义序列才能准确命中目标特殊字符。

特殊字符转义映射表 #

所有涉及的控制字符在PySpark字符串、正则中的标准表示如下：

• 字段首尾包裹的退格符（BS）：ASCII编码8，转义写法为\x08
• 字段内的换行符（LF）：ASCII编码10，转义写法为\n
• 字段内的回车符（CR）：ASCII编码13，转义写法为\r
• 列间分隔的Escape转义符（ESC）：ASCII编码27，转义写法为\x1b

如需同时匹配以上四类控制字符（对应问题描述的三类特殊字符场景），正则模式直接写为r'[\x08\n\r\x1b]'即可。

具体实现方案 #

根据文件大小和格式规整度，可以二选一实现：

方案1：内置CSV接口直接解析（大文件优先选，性能更高） #

Spark内置CSV阅读器支持自定义分隔符、引号字符，开启多行模式后可以自动识别被包裹的字段内容，不会把字段内的换行、回车误判为记录断行，配置代码如下：

from pyspark.sql import functions as F

# 按自定义控制字符规则解析文件
raw_df = spark.read \
    .option("delimiter", "\x1b")  # 指定列分隔符为ESC转义符
    .option("quote", "\x08")      # 指定字段包裹符为退格符，读取时会自动去除首尾的退格
    .option("multiLine", "true")  # 开启多行模式，自动处理字段内的换行、回车，避免断行错误
    .csv("你的文件存储路径")

注：读取后默认列名为_c0、_c1、_c2，可通过toDF("字段1名", "字段2名", "字段3名")指定自定义列名。

如果不需要保留字段内的换行、回车，可以在读取后批量清洗所有列：

# 将所有列内的换行、回车替换为空格，可根据需求调整替换值
clean_df = raw_df.select([
    F.regexp_replace(F.col(col_name), r'[\n\r]', ' ').alias(col_name) 
    for col_name in raw_df.columns
])

方案2：全量文本读取后正则清洗（适配格式不规则的小文件） #

如果文件存在格式不规范、内置CSV接口解析异常的情况，可以先将整个文件读取为纯文本，再通过正则统一处理：

from pyspark.sql import functions as F

# 读取整个文件为单列文本，避免提前被换行切分
raw_text = spark.read.text("你的文件存储路径", wholetext=True)

# 第一步：统一替换所有不需要的控制字符（退格、换行、回车、ESC）
clean_text = raw_text.withColumn(
    "value",
    F.regexp_replace(F.col("value"), r'[\x08\n\r\x1b]', '')
)

# 后续可根据清洗后的文本结构拆分字段、拆分记录即可

注意：该方案会清空所有匹配到的控制字符，如果需要保留字段内的有效换行，请调整正则匹配规则，仅替换非字段内容的控制字符。

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