如何在PySpark中实现表头无引号、数据行带双引号导出CSV

方案一：PySpark批量处理（适合大数据量） #

通过自定义数据格式拼接带引号的字段，再单独生成表头并合并文件：

from pyspark.sql import SparkSession
from pyspark.sql.functions import concat_ws, lit, col
import os
import shutil

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

# 示例DataFrame（替换为你的实际数据）
data = [("DR1", "Dr2", "DR3"), ("DR2", "DR", "DR4")]
df = spark.createDataFrame(data, ["head1", "head2", "head3"])

# 1. 处理数据行：每个字段包裹双引号，用逗号拼接成单行字符串
quoted_data_df = df.select(
    concat_ws(
        ",",
        *[lit('"') + col(c).cast("string") + lit('"') for c in df.columns]
    ).alias("line")
)

# 2. 导出数据行（不带表头，合并为单个文件方便后续处理）
temp_data_dir = "./temp_csv_data"
quoted_data_df.coalesce(1).write.mode("overwrite").text(temp_data_dir)

# 3. 生成表头字符串
header = ",".join(df.columns)

# 4. 找到Spark生成的分区数据文件
part_files = [f for f in os.listdir(temp_data_dir) if f.startswith("part-")]
if part_files:
    # 合并表头和数据到最终文件
    final_csv_path = "./output.csv"
    with open(final_csv_path, "w") as final_file:
        final_file.write(header + "\n")
        with open(os.path.join(temp_data_dir, part_files[0]), "r") as part_file:
            final_file.write(part_file.read())
    
    # 清理临时目录
    shutil.rmtree(temp_data_dir)

注意事项： #

• coalesce(1)会将所有数据合并到一个分区，适合小到中等数据量；大数据量建议保留多分区，后续合并多个part文件即可。
• 若字段本身包含双引号，可提前用regexp_replace转义（比如替换为""），避免CSV格式错误。

方案二：原生Python本地处理（仅适合小数据量） #

如果数据量不大，可将数据拉到Driver节点后用原生Python处理：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SmallDataCSVExport").getOrCreate()

# 示例DataFrame
data = [("DR1", "Dr2", "DR3"), ("DR2", "DR", "DR4")]
df = spark.createDataFrame(data, ["head1", "head2", "head3"])

# 生成表头
header = ",".join(df.columns)

# 收集数据并处理成带双引号的行
rows = df.collect()
quoted_rows = [",".join([f'"{str(val)}"' for val in row]) for row in rows]

# 写入文件
with open("./output.csv", "w") as f:
    f.write(header + "\n")
    f.write("\n".join(quoted_rows))

注意事项： #

• collect()会将全量数据加载到Driver节点内存，数据量大时可能引发内存溢出，仅适用于小数据集。

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