如何在PySpark中删除多余分号？解决CSV分号分隔失效问题

优先排查读取阶段的问题 #

sep参数未生效通常是因为CSV文件格式和指定参数不匹配，比如字段被引号包裹、存在转义字符，或是实际分隔符并非你设定的半角分号。

1. 修正读取参数确保分隔符生效 #

如果CSV字段被双引号包裹，PySpark默认会忽略引号内的分隔符，此时需显式指定引号相关参数：

input_df = spark.read.csv(
    tables_map[k],
    header=True,
    sep=";",
    encoding="iso-8859-1",
    quote='"',  # 指定包裹字段的引号字符
    escape='"',  # 指定字段内引号的转义字符
    ignoreLeadingWhiteSpace=True,  # 可选：忽略字段前的空格
    ignoreTrailingWhiteSpace=True   # 可选：忽略字段后的空格
)

2. 先确认CSV的实际内容 #

若不确定文件格式，可先读取几行原始文本验证：

# 读取前5行原始内容
sample_lines = spark.read.text(tables_map[k]).limit(5).collect()
for line in sample_lines:
    print(line.value)

比如你可能会发现实际分隔符是; （分号加空格），这时只需把sep改成"; "即可。

若数据已读取，直接清洗字段中的分号 #

如果已经完成读取操作，直接对现有字段做清洗处理：

1. 批量清洗所有字符串字段 #

from pyspark.sql.functions import regexp_replace, col

# 获取所有字符串类型的字段
string_columns = [col_name for col_name, dtype in input_df.dtypes if dtype == "string"]

# 移除每个字符串字段中的所有分号
cleaned_df = input_df.select(
    *[regexp_replace(col(col_name), ";", "").alias(col_name) if col_name in string_columns else col_name 
      for col_name in input_df.columns]
)

2. 针对特定字段精准处理 #

如果只有部分字段存在问题，直接单独处理目标字段：

from pyspark.sql.functions import regexp_replace, col

cleaned_df = input_df \
    .withColumn("LastName", regexp_replace(col("LastName"), ";", "")) \
    .withColumn("Gender", regexp_replace(col("Gender"), ";", "")) \
    .withColumn("Birth", regexp_replace(col("Birth"), ";", ""))

执行以上代码后，就能得到你需要的无分号的正确格式数据。

内容的提问来源于stack exchange，提问作者Javier León