PySpark读取ADLS Gen2列数不一致CSV文件方案求解

Spark内置CSV解析器默认逻辑为：未指定schema时以首行列数为基准截断超长列；手动指定schema时默认过滤列数与schema不匹配的损坏记录，这是出现截断、丢行问题的核心原因。以下是可直接落地的解决方法：

方案1：原生PySpark参数配置读取（适合无复杂转义字符的常规CSV） #

你已经提前知晓12列的列名和类型，直接按如下配置读取即可，无需额外逐行处理：

• 关闭首行表头识别：列数不足的行可能出现在首行，不能让Spark自动从首行推导表头
• 解析模式设置为PERMISSIVE：遇到列数不匹配的行时不直接丢弃
• 显式指定最大列数为12，避免超长列截断异常

示例代码：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType  # 按实际字段类型导入对应类

# 替换为你实际的12列字段名、字段类型
csv_schema = StructType([
    StructField("user_id", StringType(), nullable=True),
    StructField("user_name", StringType(), nullable=True),
    StructField("age", IntegerType(), nullable=True),
    # 按顺序补全剩余9个字段即可，总计12个
    StructField("col12", StringType(), nullable=True)
])

adls_path = "abfss://<容器名>@<ADLS Gen2账户名>.dfs.core.windows.net/<文件路径>.csv"

df = spark.read.format("csv") \
    .schema(csv_schema) \
    .option("header", "false") \
    .option("mode", "PERMISSIVE") \
    .option("delimiter", ",") \  # 替换为实际分隔符，比如制表符\t
    .option("maxColumns", "12") \
    .load(adls_path)

注意：如果文件中存在单独的表头行（即存储字段名的行），读取完成后手动过滤掉对应字段值等于字段名的行即可。

方案2：逐行自定义解析（兼容性最高，适合含转义字符、引号包裹字段的场景） #

如果CSV存在字段内含分隔符、换行符的复杂情况，直接绕开Spark内置CSV解析器的列数对齐逻辑，可100%避免截断、丢行问题：

1. 先把所有行按纯文本读入，每行作为单条字符串记录
2. 对每条记录做CSV解析，映射到提前定义的12列schema，列数不足时自动补null

示例代码：

from pyspark.sql.functions import from_csv, col

# 沿用上面定义的csv_schema和adls_path变量
raw_df = spark.read.text(adls_path)

# 按CSV规则解析每行文本，列数不足自动填充null
df = raw_df.select(
    from_csv(
        col("value"),
        csv_schema.simpleString(),
        {"mode": "PERMISSIVE", "delimiter": ","}
    ).alias("data")
).select("data.*")

如果CSV没有任何转义规则，分隔符不会出现在字段内容里，也可以直接用split拆分实现，性能更高：

from pyspark.sql.functions import split

# 按实际顺序填写12列的列名
target_cols = ["col1","col2","col3","col4","col5","col6","col7","col8","col9","col10","col11","col12"]

df = raw_df.select(
    [split(col("value"), ",").getItem(idx).alias(target_cols[idx]) for idx in range(12)]
)

ADF映射数据流适配配置 #

如果用ADF读取文件，调整两个配置即可实现相同效果：

• 在CSV数据集配置中关闭「第一行作为表头」选项
• 在源转换的架构设置中手动导入12列的自定义schema，将「行内容不匹配时的处理规则」设置为「继续」，不要选择「忽略不匹配行」或「任务失败」
• 配置完成后，列数不足的行会自动在末尾填充null，不会被过滤，超长列会按顺序映射到前12个字段。

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