PySpark CSV写入时在字符串列中保留字面量 字符的策略

当使用pyspark将包含 ` ` 字符的字符串列写入csv文件时，这些字符常被误解释为实际的换行符，导致数据被错误地分割到多行。本教程将介绍一种有效策略，通过自定义用户定义函数（udf）在写入前将字符串中的 ` ` 和 ` ` 字符转换为其字面量转义表示 `\r` 和 `\n`，从而确保数据在csv中以单行完整保留。

问题分析：CSV写入与特殊字符

在数据处理中，字符串列中包含回车符（ ）和换行符（ ）是常见情况。然而，当这些数据通过PySpark写入CSV文件时，默认的CSV写入器通常会将 和 解释为记录的分隔符或行内换行符，即使字段被引用（quoteAll=True）也可能无法达到预期的效果。例如，字符串 "ABCD DEFG XYZ" 在写入CSV后，可能被错误地解析为：

"ABCD
DEFG
XYZ"

这导致原始的单行记录被拆分为多行，从而破坏了数据的完整性和下游系统的解析逻辑。

问题的核心在于对 和 \n 的理解：

• 是一个特殊的非打印字符，代表“换行”。在许多文本编辑器和CSV解析器中，它被视为行的结束。
• \n 是两个可打印的字符：一个反斜杠 和一个字母 n。它代表字符串中字面意义上的反斜杠和 n，而不是换行符。

为了在CSV中保留 的字面量形式，我们需要在写入前将字符串中的实际 和 字符替换为它们的转义表示 \r 和 \n。

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

解决此问题的方法是在将数据写入CSV之前，对包含 和 的字符串列进行预处理。具体来说，我们需要编写一个PySpark用户定义函数（UDF），将字符串中的所有 替换为 \r，并将所有 替换为 \n。这样，当PySpark将处理后的字符串写入CSV时，它会将其视为普通的文本序列，而不是换行指令。

实现步骤：

1. 定义UDF： 创建一个Python函数，接收一个字符串作为输入，并返回一个经过字符替换的新字符串。
2. 注册UDF： 使用 pyspark.sql.functions.udf 将Python函数注册为PySpark UDF。
3. 应用UDF： 将UDF应用到目标DataFrame的字符串列上，生成一个新的列或更新现有列。
4. 写入CSV： 将处理后的DataFrame写入CSV文件。

代码示例

以下是完整的PySpark代码示例，演示了如何实现上述解决方案：

怪兽AI数字人

数字人短视频创作，数字人直播，实时驱动数字人

44

查看详情

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

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

# 示例数据：包含 
 的字符串
data_with_newlines = "ABCD 
 DEFG 
 XYZ"
df = spark.createDataFrame([(data_with_newlines,)], schema='col: string')

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

# 定义一个Python函数来转义 
 和 

def escape_newline_chars(s):
    if s is None:
        return None
    # 将实际的 
 替换为字面量 \r
    # 将实际的 
 替换为字面量 \n
    return s.replace('
', '\r').replace('
', '\n')

# 注册UDF
# 确保指定返回类型，这里是StringType
format_string_udf = udf(escape_newline_chars, StringType())

# 应用UDF到目标列
df_processed = df.withColumn('col', format_string_udf('col'))

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

# 将处理后的DataFrame写入CSV
# 为了简化，这里写入到单个文件，并包含header
output_path = "csv_newline_escaped"
df_processed.coalesce(1).write.csv(output_path, header=True, mode="overwrite")

print(f"
数据已成功写入到 '{output_path}' 目录下的CSV文件。")

# 停止SparkSession
spark.stop()

# 验证CSV文件内容（在终端执行）
# $ cat csv_newline_escaped/part-0000*.csv
# col
# "ABCD 
 DEFG 
 XYZ"

代码说明：

• escape_newline_chars 函数负责核心的替换逻辑。它将所有 和 字符分别替换为 \r 和 \n。
• udf(escape_newline_chars, StringType()) 将这个Python函数注册为一个PySpark UDF，并明确指定了其返回类型为 StringType，这对于性能和类型安全很重要。
• df.withColumn('col', format_string_udf('col')) 将UDF应用到名为 col 的列上，生成一个包含转义字符的新DataFrame。
• df_processed.coalesce(1).write.csv(...) 将处理后的数据写入CSV。coalesce(1) 用于将所有数据写入一个CSV文件（方便验证），header=True 添加列头，mode="overwrite" 确保每次运行时都会覆盖旧的输出。

输出验证

执行上述代码后，你可以在输出目录 csv_newline_escaped 中找到生成的CSV文件。使用命令行工具（如 cat）查看文件内容，你会发现 字符被正确地保留为字面量：

$ cat csv_newline_escaped/part-0000*.csv
col
ABCD 
 DEFG 
 XYZ

请注意，尽管在 cat 命令的输出中，ABCD DEFG XYZ 看起来没有被双引号包裹，但实际上CSV写入器通常会在字段包含特殊字符时（如空格、逗号或这里我们期望的 ）自动添加引号。如果你的下游系统期望双引号包裹，且原始数据包含这些转义字符，PySpark的CSV写入器通常会正确处理。

注意事项

1. 下游系统兼容性： 这种方法改变了字符串的实际内容。下游系统在读取此CSV文件时，如果需要原始的 行为（即将其解释为换行），则需要进行反向转换，即将 \r 转换回 ，\n 转换回 。
2. UDF性能： 尽管UDF在PySpark中非常灵活，但通常比内置函数效率低。对于大规模数据，如果性能是关键考量，应优先考虑使用PySpark内置函数。然而，对于这种简单的字符串替换操作，UDF的性能开销通常在可接受范围内。
3. CSV解析器行为： 不同的CSV解析器对包含 的字段有不同的处理方式。本教程的方法确保了 作为字面量字符存在于CSV中，避免了被解析为换行。如果你的目标是让CSV解析器正确处理一个包含多行内容的带引号字段，那么通常只需要 quoteAll=True 选项。但本问题场景表明，即使使用 quoteAll=True，仍然希望 以字面量形式出现，因此此UDF方法是更直接的解决方案。

总结

通过在PySpark中定义并应用一个简单的UDF，我们能够有效地将字符串列中的实际回车符和换行符转换为其转义表示。这种预处理策略确保了在将DataFrame写入CSV文件时，这些特殊字符不会被误解释为行分隔符，从而保证了数据的完整性和下游系统对单行记录的正确解析。在实施此方案时，务必考虑下游数据消费方对这种数据格式变化的兼容性。

以上就是PySpark CSV写入时在字符串列中保留字面量 字符的策略的详细内容，更多请关注php中文网其它相关文章！