Spark.read读取CSV指定Schema后数据未匹配对应列问题

代码实现 #

from pyspark.sql.types import StructType, StructField, StringType, DateType, DoubleType

# 定义自定义Schema
schema = StructType([
    StructField('TRANSACTION', StringType(), True),
    StructField('FROM', StringType(), True),
    StructField('TO', StringType(), True),
    StructField('DA_RATE', DateType(), True),
    StructField('CURNCY_F', StringType(), True),
    StructField('CURNCY_T', StringType(), True)
])

# 读取CSV文件
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "|") \
    .schema(schema) \
    .load("abfss://xyz@abc.dfs.core.windows.net/my/2024-04-10/abc_2*.csv")

CSV文件内容 #

DA_RATE|CURNCY_F|CURNCY_T
2024-02-26|AAA|MMM
2024-02-26|AAA|NNN
2024-02-26|BBB|YYY
2024-02-26|CCC|KKK
2024-02-27|DDD|SSS

实际输出 #

期望输出 #

问题原因 #

Spark读取CSV时，若显式指定了Schema，会完全忽略CSV的表头信息，仅按照Schema中定义的列顺序匹配CSV的列。你的CSV只有3列，而Schema定义了6列，因此CSV的第一列被映射到Schema的第一列TRANSACTION，第二列到FROM，第三列到TO，剩余的DA_RATE、CURNCY_F、CURNCY_T列因无对应数据，全部显示为null。

解决方案 #

方案1：调整Schema列顺序，匹配CSV列顺序 #

将CSV中存在的列放到Schema最前面，读取后再调整列顺序到期望格式：

from pyspark.sql.types import StructType, StructField, StringType, DateType
from pyspark.sql import functions as F

# 调整Schema列顺序，先匹配CSV的列
schema = StructType([
    StructField('DA_RATE', DateType(), True),
    StructField('CURNCY_F', StringType(), True),
    StructField('CURNCY_T', StringType(), True),
    StructField('TRANSACTION', StringType(), True),
    StructField('FROM', StringType(), True),
    StructField('TO', StringType(), True)
])

# 读取CSV
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "|") \
    .schema(schema) \
    .load("abfss://xyz@abc.dfs.core.windows.net/my/2024-04-10/abc_2*.csv")

# 调整列顺序到期望的格式
df = df.select("TRANSACTION", "FROM", "TO", "DA_RATE", "CURNCY_F", "CURNCY_T")

方案2：先读取CSV，再补充缺失列并调整顺序 #

不指定Schema，先读取CSV后再添加缺失列，同时转换数据类型：

from pyspark.sql.types import StringType, DateType
from pyspark.sql import functions as F

# 读取CSV，自动识别表头
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "|") \
    .option("inferSchema", "false") \
    .load("abfss://xyz@abc.dfs.core.windows.net/my/2024-04-10/abc_2*.csv")

# 转换DA_RATE为日期类型
df = df.withColumn("DA_RATE", F.col("DA_RATE").cast(DateType()))

# 添加缺失的列并设置为null
df = df.withColumn("TRANSACTION", F.lit(None).cast(StringType())) \
       .withColumn("FROM", F.lit(None).cast(StringType())) \
       .withColumn("TO", F.lit(None).cast(StringType()))

# 调整列顺序
df = df.select("TRANSACTION", "FROM", "TO", "DA_RATE", "CURNCY_F", "CURNCY_T")

方案3：使用列名映射（适合列名不匹配的场景） #

通过selectExpr直接映射列并补充缺失字段：

from pyspark.sql import functions as F

# 读取CSV
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("delimiter", "|") \
    .load("abfss://xyz@abc.dfs.core.windows.net/my/2024-04-10/abc_2*.csv")

# 映射列并添加缺失列
df = df.selectExpr(
    "null as TRANSACTION",
    "null as FROM",
    "null as TO",
    "cast(DA_RATE as date) as DA_RATE",
    "CURNCY_F",
    "CURNCY_T"
)

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