如何在PySpark CSV输出中保留字符串中的\r\n（不换行）-Python教程-PHP中文网

如何在PySpark CSV输出中保留字符串中的（不换行）

碧海醫心

发布： 2025-10-13 10:29:15

原创

416人浏览过

如何在PySpark CSV输出中保留字符串中的
（不换行）

本文详细介绍了pyspark在将包含换行符（` `和` `）的字符串写入csv文件时，如何避免这些字符被解释为实际的换行，从而导致数据记录被错误地分割。核心解决方案是利用pyspark udf将字符串中的` `和` `字符转义为`\r`和`\n`，确保它们作为字面量被写入，从而在下游系统中正确解析。

在数据处理流程中，我们经常需要将PySpark DataFrame中的数据写入CSV文件。然而，当字符串列中包含回车符（）或换行符（）时，PySpark的默认CSV写入行为可能会导致这些特殊字符被解释为实际的行终止符，从而将单个逻辑记录拆分成多行，这通常不是我们期望的结果。例如，一个包含"ABCD DEFG XYZ"的字符串，在写入CSV后，可能会在文本编辑器中显示为：

"ABCD
DEFG
XYZ"

登录后复制

这与我们希望保留原始字面量"ABCD DEFG XYZ"的意图相悖。即使尝试使用quoteAll=True或escape等选项，PySpark的CSV写入器通常仍会将和视为内部换行符。

问题根源分析

问题的核心在于对特殊字符的解释。在Python或许多编程语言中，是一个单一的非打印字符，代表“新行”。而我们希望在CSV中看到的\n则是两个可打印的字符：一个反斜杠和一个字母n。PySpark的CSV写入器在处理字符串时，会识别并处理、等特殊字符，而不是将其原样输出为字面量的和n。

>>> len('
')   # 单个非打印字符 'new line'
1
>>> len('\n')  # 两个可打印字符 '' (转义) 和 'n'
2

登录后复制

为了解决这个问题，我们需要在将数据写入CSV之前，显式地将字符串中的和字符“转义”，即将其替换为它们的字面量表示\r和\n。

解决方案：使用UDF进行字符转义

最直接有效的方法是使用PySpark的用户自定义函数（UDF）来预处理包含潜在换行符的字符串列。UDF允许我们定义自定义的Python函数，并在DataFrame的列上应用它。

1. 定义UDF

首先，我们需要导入udf函数，并定义一个Python函数，该函数接收一个字符串，然后将字符串中的替换为\r，将替换为\n。

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 定义一个UDF来转义回车和换行符
def escape_newlines(s):
    if s is None:
        return None
    return s.replace('
', '\r').replace('
', '\n')

# 将Python函数注册为PySpark UDF
format_string_udf = udf(escape_newlines, StringType())

登录后复制

这里我们明确指定了UDF的返回类型为StringType()，这是一个良好的实践，有助于Spark进行优化。

2. 应用UDF到DataFrame列

接下来，我们将这个UDF应用到包含问题字符串的DataFrame列上。

BetterYeah AI

基于企业知识库构建、训练AI Agent的智能体应用开发平台，赋能客服、营销、销售场景 -BetterYeah

110

查看详情

from pyspark.sql import SparkSession

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

# 示例数据
s = "ABCD 
 DEFG 
 XYZ"
df = spark.createDataFrame(data=[(s,)], schema='col: string')

print("原始DataFrame内容:")
df.show(truncate=False)

# 应用UDF转义字符串列
df_escaped = df.withColumn('col', format_string_udf('col'))

print("应用UDF后的DataFrame内容:")
df_escaped.show(truncate=False)

登录后复制

运行上述代码，你会看到df_escaped中的col列现在显示为"ABCD \r\n DEFG \r\n XYZ"，这意味着和已经被成功转义。

3. 写入CSV文件

现在，我们可以将处理后的DataFrame写入CSV文件。由于我们已经将和转义成了字面量，PySpark的CSV写入器将不再将其视为换行符。

# 写入CSV文件
output_path = "csv_newline_escaped"
df_escaped.write.mode("overwrite").csv(output_path, header=True)

print(f"
数据已写入到 {output_path} 目录。")

# 停止SparkSession
spark.stop()

登录后复制

我们使用了mode("overwrite")以防目录已存在，并设置header=True以包含列名。

4. 验证输出

为了验证CSV文件内容是否符合预期，我们可以使用命令行工具（如cat）来查看生成的文件。

# 在Linux/macOS环境下，进入spark-submit运行目录或指定的output_path
# 假设 output_path 是 'csv_newline_escaped'
$ cat csv_newline_escaped/part-0000*.csv
col
ABCD 
 DEFG 
 XYZ

登录后复制

可以看到，CSV文件中的字符串完全保留了\r\n的字面量形式，从而实现了单行记录的正确输出。

注意事项与最佳实践

性能考量： UDF在PySpark中通常会引入性能开销，因为它需要在Python解释器和JVM之间进行数据序列化和反序列化。对于大规模数据集，如果性能成为瓶颈，可以考虑使用Spark SQL内置函数如regexp_replace来完成类似的替换，尽管它可能在处理多个不同字符时稍微复杂一些。例如：
```
from pyspark.sql.functions import regexp_replace
df_sql_escaped = df.withColumn('col', regexp_replace('col', '
', '\r')) 
                   .withColumn('col', regexp_replace('col', '
', '\n'))
```
登录后复制
这种方式通常比UDF更高效。
下游系统兼容性： 确保接收此CSV文件的下游系统能够正确地解析\r和\n作为转义的字面量，并将其转换回实际的换行符（如果这是其预期行为）。不同的解析器可能有不同的转义规则。
其他特殊字符： 如果字符串中还包含其他需要保留字面量的特殊字符（例如CSV分隔符、引号等），也需要考虑对其进行类似的转义处理。

总结

当PySpark DataFrame中的字符串列包含或等换行符，且希望在写入CSV文件时这些字符作为字面量\r和\n而非实际换行符保留时，最可靠的方法是使用PySpark UDF（或Spark SQL函数）在写入前对这些字符进行转义。通过将替换为\r，替换为\n，可以确保CSV文件中的每条记录都保持其预期的单行结构，从而避免数据解析错误，并满足下游系统对数据格式的严格要求。

以上就是如何在PySpark CSV输出中保留字符串中的（不换行）的详细内容，更多请关注php中文网其它相关文章！