如何从PySpark DataFrame获取原始CSV行字符串处理校验失败数据？

好问题！其实默认情况下，PySpark在解析CSV生成DataFrame时，会把原始的文本行转换成结构化的字段，并不会保留原始的CSV字符串——所以直接从已解析的DataFrame里拿原始行是做不到的。不过我们可以换个思路，在加载阶段就把原始行和解析后的字段一起保留，这样后续校验失败时就能直接导出原始行了，完全不需要回头去原文件提取。

具体实现步骤 #

1. 先读取原始文本行，再解析CSV
   先用spark.read.text()读取整个CSV文件的所有原始行，然后再对这些行进行解析，把原始行和解析后的字段同时存在DataFrame中。这样既保留了结构化数据用于校验，又能随时拿到原始的CSV字符串。

   示例代码（假设是逗号分隔的简单CSV，带header）：

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import split, col
   
   spark = SparkSession.builder.appName("CSVValidationWithRawLine").getOrCreate()
   
   # 读取所有原始行
   raw_lines_df = spark.read.text("your_input.csv")
   
   # 提取header并过滤掉数据行中的header（避免解析第一行）
   header_row = raw_lines_df.first()[0]
   data_lines_df = raw_lines_df.filter(col("value") != header_row)
   
   # 解析CSV行到结构化字段（根据你的CSV格式调整分隔符、字段数）
   parsed_df = data_lines_df.withColumn(
       "parsed_fields",
       split(col("value"), ",")
   ).select(
       col("value").alias("raw_csv_line"),  # 保留原始行
       col("parsed_fields")[0].alias("user_id"),
       col("parsed_fields")[1].alias("username"),
       col("parsed_fields")[2].alias("email")
       # 其他字段依次类推
   )
   

   如果你的CSV格式比较复杂（比如带引号、转义字符），建议结合csv()的解析逻辑来处理，而不是简单的split()——可以先把原始行转成临时的CSV流来解析：

   from pyspark.sql.functions import from_csv, schema_of_csv
   
   # 自动推断CSV schema（也可以手动定义）
   csv_schema = schema_of_csv(header_row, sep=",", quote='"')
   
   # 用from_csv解析原始行
   parsed_df = data_lines_df.withColumn(
       "parsed_data",
       from_csv(col("value"), csv_schema, {"sep": ",", "quote": '"'})
   ).select(
       col("value").alias("raw_csv_line"),
       col("parsed_data.*")  # 展开所有结构化字段
   )
   

2. 校验并导出失败的原始行
   现在你的DataFrame里既有结构化字段用于校验，又有raw_csv_line存储原始行。接下来只需添加校验逻辑，过滤出校验失败的行，直接导出raw_csv_line即可：

   # 示例校验逻辑：user_id不能为null，email必须包含@
   validated_df = parsed_df.withColumn(
       "is_valid",
       col("user_id").isNotNull() & col("email").contains("@")
   )
   
   # 过滤出校验失败的行，写入文件
   invalid_lines_df = validated_df.filter(col("is_valid") == False)
   invalid_lines_df.select("raw_csv_line").write.mode("overwrite").text("invalid_csv_lines.txt")
   

补充说明 #

如果已经用spark.read.csv()直接加载了CSV，没有保留原始行，那确实没办法从已有的DataFrame中恢复原始的CSV字符串——因为解析过程中Spark会丢弃原始的格式细节（比如引号、转义符、换行符等）。这种情况下如果要拿原始行，只能回到原文件提取，但分布式环境下DataFrame的行顺序和原文件的行号不一定对应（Spark会打乱分区处理），所以这种方式可靠性不高。

综上，最稳妥的方式还是在加载阶段就保留原始行，这样后续的校验和导出都能直接基于DataFrame完成，完全符合你的需求。

内容的提问来源于stack exchange，提问作者gCoh