PySpark读取CSV时修改列数据类型及StructType顺序问题解决

当使用StructType定义Schema时，如果列顺序和CSV文件的列顺序不匹配，Spark会默认按Schema的顺序映射CSV列（忽略列名），导致列名和数据错位。以下是几种实用的解决方法：

方法一：先读取数据再修改指定列类型（适合少量列修改） #

先完整读取CSV保留原列名和顺序，再对需要改类型的列单独转换，操作简单直接：

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType, FloatType

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

# 读取CSV，启用表头识别并自动推断初始Schema
df = spark.read.csv("your_file.csv", header=True, inferSchema=True)

# 按需修改指定列的类型，比如把columnB转为浮点型，columnD转为字符串型
df_modified = df.withColumn("columnB", df["columnB"].cast(FloatType())) \
                .withColumn("columnD", df["columnD"].cast(StringType()))

# 验证结果
df_modified.printSchema()

方法二：动态构建与CSV表头顺序一致的Schema（适合多列自定义类型） #

先获取CSV的表头顺序，再根据需求动态生成匹配顺序的Schema，既保证列顺序正确，又能自定义指定列的类型：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, FloatType

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

# 读取表头行，获取原始列顺序
header_df = spark.read.csv("your_file.csv", header=True, maxRows=0)
column_order = header_df.columns

# 定义需要修改类型的列映射（键为列名，值为目标类型）
type_mapping = {
    "columnB": FloatType(),
    "columnD": StringType()
}

# 按原始列顺序构建Schema，未指定的列用默认IntegerType
schema_fields = []
for col_name in column_order:
    data_type = type_mapping.get(col_name, IntegerType())
    schema_fields.append(StructField(col_name, data_type, nullable=True))

custom_schema = StructType(schema_fields)

# 使用自定义Schema读取CSV
df = spark.read.csv("your_file.csv", header=True, schema=custom_schema)

df.printSchema()

方法三：启用列名匹配模式（允许Schema列顺序与CSV不同） #

如果必须使用自定义Schema且列顺序和CSV不一致，只要确保CSV有表头，读取时设置header=True，Spark会自动按列名匹配Schema字段，而不是顺序：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, FloatType

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

# 自定义Schema，列顺序可以和CSV不同，但列名必须完全对应
custom_schema = StructType([
    StructField("columnD", StringType(), nullable=True),
    StructField("columnB", FloatType(), nullable=True),
    StructField("columnA", IntegerType(), nullable=True),
    StructField("columnC", IntegerType(), nullable=True),
    StructField("columnE", IntegerType(), nullable=True)
])

# 读取时启用表头识别，Spark会按列名匹配Schema
df = spark.read.csv("your_file.csv", header=True, schema=custom_schema)

# 最终DataFrame的列顺序会和CSV原始顺序一致
df.printSchema()

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