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

聖光之護

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

原创

247人浏览过

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

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

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

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

"ABCD
DEFG
XYZ"

登录后复制

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

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

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

1. 定义并注册UDF

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

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
<h1>定义一个Python函数，将 
 替换为 
，将 
 替换为 
</h1><p>def escape_newlines(s):
if s is None:
return None</p><h1>注意：这里是替换实际的换行符 '
' 和 '
'</h1><pre class="brush:php;toolbar:false;"># 替换成它们的转义字符串 '\r' 和 '\n'
return s.replace('
', '\r').replace('
', '\n')

登录后复制

将Python函数注册为PySpark UDF

指定返回类型为StringType

format_string_udf = udf(escape_newlines, StringType())

2. 应用UDF到DataFrame列

接下来，我们将这个UDF应用到包含问题字符串的DataFrame列上。以下是一个示例，展示如何创建一个包含换行符的DataFrame，并应用UDF进行转换：

from pyspark.sql import SparkSession
<h1>初始化SparkSession</h1><p>spark = SparkSession.builder.appName("EscapeNewlinesInCSV").getOrCreate()</p><h1>示例数据</h1><p>s = "ABCD 
 DEFG 
 XYZ"
df = spark.createDataFrame(data=[(s,)], schema='col: string')</p><p>print("原始DataFrame内容 (show()可能直接显示为多行，但内部仍是一个字符串):")
df.show(truncate=False)</p><h1>示例输出可能看起来像：</h1><h1>+-----------------------+</h1><h1>|col                    |</h1><h1>+-----------------------+</h1><h1>|ABCD</h1><h1>DEFG</h1><h1>XYZ|</h1><h1>+-----------------------+</h1><h1>应用UDF转换列</h1><p>df_processed = df.withColumn('col', format_string_udf('col'))</p>
                    <div class="aritcle_card">
                        <a class="aritcle_card_img" href="/ai/914">
                            <img src="https://img.php.cn/upload/ai_manual/000/000/000/175679997026132.png" alt="灵感PPT">
                        </a>
                        <div class="aritcle_card_info">
                            <a href="/ai/914">灵感PPT</a>
                            <p>AI灵感PPT - 免费一键PPT生成工具</p>
                            <div class="">
                                <img src="/static/images/card_xiazai.png" alt="灵感PPT">
                                <span>226</span>
                            </div>
                        </div>
                        <a href="/ai/914" class="aritcle_card_btn">
                            <span>查看详情</span>
                            <img src="/static/images/cardxiayige-3.png" alt="灵感PPT">
                        </a>
                    </div>
                <p>print("
处理后的DataFrame内容 (show()显示为字面量):")
df_processed.show(truncate=False)</p><h1>+-----------------------+</h1><h1>|col                    |</h1><h1>+-----------------------+</h1><h1>|ABCD 
 DEFG 
 XYZ|</h1><h1>+-----------------------+</h1><p>

登录后复制

在 df_processed.show(truncate=False) 的输出中，您会看到 ` ` 已经作为字面量显示在字符串中，而不是导致行中断。

3. 写入CSV文件并验证

最后，我们将处理后的DataFrame写入CSV文件。此时，由于 ` ` 和 ` ` 已经被替换为 `\r` 和 `\n`，PySpark将不再将其解释为行分隔符。

# 将处理后的DataFrame写入CSV文件
output_path = "csv_newline_escaped"
# 为了避免重复运行出错，先删除旧目录
import shutil
shutil.rmtree(output_path, ignore_errors=True)
<p>df_processed.write.csv(output_path, header=True, mode="overwrite")</p><p>print(f"
CSV文件已写入到: {output_path}")</p><h1>验证CSV文件内容（在Linux/macOS系统上可以使用cat命令）</h1><h1>您可能需要根据实际的part-xxxx.csv文件名进行调整</h1><h1>示例命令和输出：</h1><h1>$ cat csv_newline_escaped/part-0000*.csv</h1><h1>col</h1><h1>"ABCD 
 DEFG 
 XYZ"</h1><p>

登录后复制

打开生成的CSV文件（例如，使用文本编辑器或命令行 cat），您会发现 "ABCD DEFG XYZ" 完整地保留在一行中，其中的 ` ` 是字面量，而不是实际的换行符。

注意事项与最佳实践

UDF性能：Python UDF在PySpark中通常比内置函数效率低，因为数据需要在JVM和Python进程之间序列化和反序列化。对于大规模数据，如果性能成为瓶颈，可以考虑使用Pandas UDF（Vectorized UDFs）或尝试寻找Spark SQL内置函数（尽管对于这种精确的转义需求可能没有直接的内置函数）。
下游系统兼容性：确保接收此CSV文件的下游系统或应用程序能够正确解析 `\r` 和 `\n` 字面量。它们可能需要进行反向的转义处理，将 `\r` 转换回 ` `，`\n` 转换回 ` `。
CSV选项：虽然本教程中的UDF是核心解决方案，但其他CSV写入选项（如 quoteAll=True, delimiter='|'）对于生成格式良好的CSV文件仍然重要。quoteAll=True 确保所有字段都被引号包围，有助于处理包含逗号等特殊字符的字段，但它不能解决内部换行符的问题。
空值处理：在UDF中增加了对 None 值的处理，确保在列中存在空值时不会引发错误。