Azure Synapse中PySpark 3.5.5用CharType读CSV报错咨询

方案1：先以StringType读取，再转换为CharType #

由于Spark CSV数据源暂不支持直接用CharType schema读取，可先将字段定义为StringType完成读取，再逐个转换为目标CharType：

from pyspark.sql.types import StructType, StructField, StringType, CharType

# 定义临时StringType schema
temp_schema = StructType([
    StructField("col1", StringType(), nullable=True),
    StructField("col2", StringType(), nullable=True),
    # 按需求添加其他字段
])

# 读取CSV文件
raw_df = spark.read.csv("your/csv/file/path", schema=temp_schema, header=True)

# 转换为目标CharType schema
final_df = raw_df.select(
    raw_df.col1.cast(CharType(10)).alias("col1"),
    raw_df.col2.cast(CharType(5)).alias("col2"),
    # 其他字段依次执行cast转换
)

方案2：通过Spark SQL建表定义CharType，再加载CSV #

利用Spark SQL的DDL语法直接定义包含CharType的外部表，再读取表数据：

CREATE TABLE IF NOT EXISTS qcew_table (
    col1 CHAR(10),
    col2 CHAR(5),
    -- 补充其他字段的CharType定义
)
USING CSV
OPTIONS (
    path "your/csv/file/path",
    header "true",
    delimiter ","
)

之后在PySpark中读取该表：

final_df = spark.table("qcew_table")

方案3：补全固定长度后转换为CharType #

CharType要求固定长度（不足补空格），可先读取为StringType，对字段补空格至指定长度后再转换：

from pyspark.sql.functions import lpad
from pyspark.sql.types import CharType

# 直接读取CSV（自动推断或指定StringType schema）
raw_df = spark.read.csv("your/csv/file/path", header=True)

# 补空格并转换为CharType
final_df = raw_df.withColumn("col1", lpad(col("col1"), 10, " ").cast(CharType(10))) \
                 .withColumn("col2", lpad(col("col2"), 5, " ").cast(CharType(5)))

问题说明 #

虽然CharType属于Spark的AtomicType，但CSV数据源的内置读取逻辑未实现直接映射CharType的处理，因此直接用包含CharType的schema读取会触发报错。上述方案均无需启用spark.sql.legacy.charVarcharAsString=true，可保留CharType的原生特性。

内容的提问来源于stack exchange，提问作者Vijay Tripathi