PySpark CSV写入：保留字符串中的 \r\n 字面量而非换行符

聖光之護

发布时间：2025-10-13 10:18:42

271人浏览过

来源于php中文网

原创

$PySpark CSV写入：保留字符串中的 \\r\\n 字面量而非换行符$

当使用pyspark将包含 `\r\n`（回车换行符）的字符串列写入csv文件时，pyspark默认会将其解释为实际的行分隔符，导致数据被错误地拆分成多行。本教程将详细介绍如何通过定义一个pyspark用户自定义函数（udf），在写入csv前将字符串中的 `\r` 和 `\n` 字符替换为其转义后的字面量 `\\r` 和 `\\n`，从而确保数据完整性，使csv文件能正确显示这些字符。

理解问题：PySpark CSV写入的默认行为

在数据处理中，字符串内包含换行符（如 `\n` 或 `\r\n`）是常见情况。例如，一个数据字段可能存储着多行文本信息，其内部结构为 "ABCD \r\n DEFG \r\n XYZ"。当我们在PySpark DataFrame中查看这样的数据时，它会显示为一个完整的字符串。然而，当尝试使用 df.write.csv() 将其写入CSV文件时，PySpark的CSV写入器会将这些内部的 `\r\n` 字符解释为CSV记录的实际行分隔符。这意味着，原本应该在一行中的数据，会被错误地拆分成多行，例如：

"ABCD
DEFG
XYZ"

这与我们期望将 `\r\n` 作为字符串的字面量而非控制字符保留在CSV文件中的行为相悖。问题的核心在于对字符 `\n`（单个非打印的换行符）和 `\\n`（两个可打印字符：反斜杠和字母n）的混淆。PySpark在写入时，会将前者直接转换为实际的换行，而我们需要的是后者。

解决方案：使用UDF预处理字符串

解决此问题的关键在于在数据写入CSV之前，对包含换行符的字符串列进行预处理。我们将使用PySpark的用户自定义函数（UDF）将字符串中实际的 `\r` 和 `\n` 字符替换为其转义后的字面量 `\\r` 和 `\\n`。这样，当PySpark写入CSV时，它看到的是字面量的反斜杠和字母，而不是需要解析的控制字符。

1. 定义并注册UDF

首先，我们需要导入 udf 函数，并定义一个Python函数来执行替换操作。这个Python函数将接收一个字符串作为输入，并返回一个处理后的字符串。

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
定义一个Python函数，将 \r 替换为 \r，将 \n 替换为 \n
def escape_newlines(s):
if s is None:
return None
注意：这里是替换实际的换行符 '\r' 和 '\n'
# 替换成它们的转义字符串 '\\r' 和 '\\n'
return s.replace('\r', '\\r').replace('\n', '\\n')
将Python函数注册为PySpark UDF
指定返回类型为StringType
format_string_udf = udf(escape_newlines, StringType())

2. 应用UDF到DataFrame列
接下来，我们将这个UDF应用到包含问题字符串的DataFrame列上。以下是一个示例，展示如何创建一个包含换行符的DataFrame，并应用UDF进行转换：
from pyspark.sql import SparkSession
初始化SparkSession
spark = SparkSession.builder.appName("EscapeNewlinesInCSV").getOrCreate()
示例数据
s = "ABCD \r\n DEFG \r\n XYZ"
df = spark.createDataFrame(data=[(s,)], schema='col: string')
print("原始DataFrame内容 (show()可能直接显示为多行，但内部仍是一个字符串):")
df.show(truncate=False)
示例输出可能看起来像：
+-----------------------+
|col                    |
+-----------------------+
|ABCD
DEFG
XYZ|
+-----------------------+
应用UDF转换列
df_processed = df.withColumn('col', format_string_udf('col'))

							
								
								
									Buildt.ai
									AI驱动的软件开发平台，可以自动生成代码片段、代码分析及其他自动化任务
								
								下载 
							
						
print("\n处理后的DataFrame内容 (show()显示为字面量):")
df_processed.show(truncate=False)
+-----------------------+
|col                    |
+-----------------------+
|ABCD \r\n DEFG \r\n XYZ|
+-----------------------+
在 df_processed.show(truncate=False) 的输出中，您会看到 `\r\n` 已经作为字面量显示在字符串中，而不是导致行中断。
3. 写入CSV文件并验证
最后，我们将处理后的DataFrame写入CSV文件。此时，由于 `\r` 和 `\n` 已经被替换为 `\\r` 和 `\\n`，PySpark将不再将其解释为行分隔符。
# 将处理后的DataFrame写入CSV文件
output_path = "csv_newline_escaped"
# 为了避免重复运行出错，先删除旧目录
import shutil
shutil.rmtree(output_path, ignore_errors=True)
df_processed.write.csv(output_path, header=True, mode="overwrite")
print(f"\nCSV文件已写入到: {output_path}")
验证CSV文件内容（在Linux/macOS系统上可以使用cat命令）
您可能需要根据实际的part-xxxx.csv文件名进行调整
示例命令和输出：
$ cat csv_newline_escaped/part-0000*.csv
col
"ABCD \r\n DEFG \r\n XYZ"
打开生成的CSV文件（例如，使用文本编辑器或命令行 cat），您会发现 "ABCD \r\n DEFG \r\n XYZ" 完整地保留在一行中，其中的 `\r\n` 是字面量，而不是实际的换行符。
注意事项与最佳实践


UDF性能：Python UDF在PySpark中通常比内置函数效率低，因为数据需要在JVM和Python进程之间序列化和反序列化。对于大规模数据，如果性能成为瓶颈，可以考虑使用Pandas UDF（Vectorized UDFs）或尝试寻找Spark SQL内置函数（尽管对于这种精确的转义需求可能没有直接的内置函数）。

下游系统兼容性：确保接收此CSV文件的下游系统或应用程序能够正确解析 `\\r` 和 `\\n` 字面量。它们可能需要进行反向的转义处理，将 `\\r` 转换回 `\r`，`\\n` 转换回 `\n`。

CSV选项：虽然本教程中的UDF是核心解决方案，但其他CSV写入选项（如 quoteAll=True, delimiter='|'）对于生成格式良好的CSV文件仍然重要。quoteAll=True 确保所有字段都被引号包围，有助于处理包含逗号等特殊字符的字段，但它不能解决内部换行符的问题。

空值处理：在UDF中增加了对 None 值的处理，确保在列中存在空值时不会引发错误。

总结
通过本文介绍的UDF方法，您可以有效地解决PySpark在写入CSV文件时，字符串列中 `\r\n` 字符被错误解析为实际换行符的问题。这种预处理策略确保了数据的完整性和一致性，使得包含特殊控制字符的字符串能够作为字面量正确地存储在CSV文件中，满足特定的数据交换需求。