在PySpark中将DataFrame写入CSV文件时，如果字符串列中包含实际的换行符（或），它们通常会被解释为行终止符，导致数据被错误地拆分成多行。本文将详细介绍如何通过自定义用户定义函数（UDF）将这些内部换行符转换为其字面量字符串表示（r和n），从而确保在CSV文件中完整保留原始字符串内容，避免数据结构被破坏。
引言：PySpark CSV写入中保留换行符的挑战
在数据处理流程中，我们经常需要将Parquet或其他格式的数据转换为CSV格式。当数据中包含字符串类型的列，且这些字符串内部含有回车符（）或换行符（）时，PySpark的CSV写入操作默认会将这些字符解释为行的分隔符。例如，一个包含”ABCD DEFG XYZ”的字符串，在写入CSV后，可能会被错误地显示为三行：
"ABCDDEFGXYZ"
这与我们期望的在CSV中保留原始字符串完整性（即”ABCD DEFG XYZ”作为一个单一字段）的目标相悖。即使尝试使用quoteAll=True或escape等选项，PySpark的CSV写入器通常仍会将实际的换行符作为物理行分隔符处理。
深入理解问题： 与 n 的区别
解决此问题的关键在于理解Python字符串中和n的根本区别：
：这是一个单个字符，代表一个“换行”控制字符。当它出现在字符串中时，通常会导致文本显示时换到下一行。其长度为1。n：这是两个字符，第一个是反斜杠（），第二个是字母n。它代表的是字面意义上的反斜杠和字母n，而不是换行符。其长度为2。
PySpark的CSV写入器在处理包含的字符串时，会将其解释为行分隔符。为了让CSV文件能够按字面意义存储，我们需要在写入前将字符串中的实际换行符和转换为它们的字面量字符串表示r和n。
解决方案：使用PySpark UDF转换换行符
我们可以通过创建一个用户定义函数（UDF）来预处理包含换行符的字符串列。这个UDF会遍历字符串中的所有实际换行符，并将它们替换为对应的字面量字符串。
1. UDF定义与原理
UDF的核心思想是将一个Python函数注册为Spark可以执行的函数。对于字符串替换，我们可以使用Python内置的str.replace()方法。
from pyspark.sql.functions import udffrom pyspark.sql.types import StringType# 定义一个UDF，用于将字符串中的实际回车和换行符替换为它们的字面量表示def format_string_for_csv(s): if s is None: return None # 将实际的回车符 '' 替换为字面量字符串 'r' # 将实际的换行符 '' 替换为字面量字符串 'n' return s.replace('', 'r').replace('', 'n')# 注册UDF，指定返回类型为StringTypeformat_string_udf = udf(format_string_for_csv, StringType())
这个format_string_for_csv函数接收一个字符串s。如果s不为None，它会执行两次替换操作：
s.replace(”, ‘r’)：将字符串中所有实际的回车符（）替换为两个字符和r。s.replace(”, ‘n’)：将字符串中所有实际的换行符（）替换为两个字符和n。
2. 应用UDF到DataFrame
假设我们有一个DataFrame df，其中包含一个名为col的字符串列，其值可能包含换行符。我们可以使用withColumn方法将UDF应用到该列，生成一个新的列（或者覆盖原有列）。
from pyspark.sql import SparkSession# 初始化SparkSessionspark = SparkSession.builder.appName("RetainNewlinesInCSV").getOrCreate()# 示例数据# 注意：这里的字符串 's' 包含实际的 和 字符s = "ABCD DEFG XYZ"df = spark.createDataFrame(data=[(s,)], schema='col: string')print("原始DataFrame内容:")df.show(truncate=False)# 输出:# +-------------------+# |col |# +-------------------+# |ABCD# DEFG# XYZ|# +-------------------+# 应用UDF转换 'col' 列df_processed = df.withColumn('col', format_string_udf('col'))print("应用UDF后的DataFrame内容:")df_processed.show(truncate=False)# 输出:# +-----------------------+# |col |# +-----------------------+# |ABCD DEFG XYZ|# +-----------------------+
从df_processed.show()的输出可以看出，现在已经显示为字面量字符串rn，这意味着它们已经被正确地转换了。
将处理后的数据写入CSV
现在，转换后的DataFrame df_processed可以安全地写入CSV文件了。由于我们已经将内部的换行符转换为字面量字符串，CSV写入器将不再将其解释为行分隔符。
# 将处理后的DataFrame写入CSV文件output_path = "csv_newline_output"df_processed.write.mode("overwrite").option("header", "true").csv(output_path)print(f"数据已成功写入到 {output_path}")
我们使用了mode(“overwrite”)来确保每次运行都能覆盖旧的输出，option(“header”, “true”)来写入列头。
结果验证
为了验证CSV文件是否正确地保留了字符串中的，我们可以查看生成的文件内容。在Linux/macOS系统上，可以使用cat命令：
# 在终端中执行以下命令（假设Spark输出目录为csv_newline_output）# 注意：PySpark通常会将CSV写入到以指定路径命名的目录下，并生成part-XXXXX.csv文件cat csv_newline_output/part-0000*.csv
预期的输出将是：
colABCD DEFG XYZ
这证明了字符串中的已被成功地作为字面量字符写入到CSV文件中，而不是导致新的行。
注意事项与最佳实践
性能考量：UDF虽然功能强大，但通常比Spark内置函数效率低。对于大规模数据，如果性能成为瓶颈，可以考虑其他方法，例如使用regexp_replace（尽管对于简单的和替换，UDF通常足够高效）。
from pyspark.sql.functions import regexp_replace# 替代UDF的方法df_processed_alt = df.withColumn('col', regexp_replace('col', '', 'r')) .withColumn('col', regexp_replace('col', '', 'n'))
这种regexp_replace链式调用通常比Python UDF性能更好。
源数据特性：如果你的源数据在读取时就已经将存储为字面量字符串rn（例如，某些系统在导出时已经做了转义），那么你就不需要执行上述UDF转换步骤。这个UDF仅适用于源数据中包含实际的或控制字符的情况。
CSV写入选项：
quoteAll=True：虽然本文的解决方案主要依赖于UDF预处理，但为了确保CSV文件的健壮性，特别是在字段可能包含分隔符或引号字符时，建议在write.csv时使用option(“quoteAll”, “true”)。这将强制所有字段都被引号包围。escape：此选项用于指定如何转义字段内的引号字符。它与处理作为行分隔符的问题无关。
数据类型：确保你的目标列是字符串类型（StringType），因为UDF是针对字符串操作设计的。
总结
通过在PySpark中定义并应用一个简单的UDF，我们可以有效地解决CSV写入时字符串内部换行符被错误解释的问题。通过将实际的和字符转换为它们的字面量字符串表示r和n，我们能够确保数据在CSV文件中以期望的单行完整形式保留，从而避免数据损坏和下游处理错误。这种方法提供了一个灵活且可控的解决方案，适用于需要精确控制CSV输出格式的场景。
以上就是PySpark CSV写入时保留字符串中换行符的策略的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1377236.html
微信扫一扫 
支付宝扫一扫 
