Synapse Notebook导入CSV至Parquet：如何强制Zip Code列为字符串类型

要解决Zip Code字段因被识别为整数丢失前导0的问题，核心是在Spark读取CSV阶段就强制将该列识别为字符串类型，同时尽量保留原有通用代码逻辑，以下是两种可行方案：

方案1：显式定义完整Schema #

直接定义包含所有列类型的Schema，将Zip Code指定为字符串类型，这种方式最稳定可靠：

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

# 根据实际CSV列结构定义Schema，示例仅列出部分列
custom_schema = StructType([
    StructField("Name", StringType(), nullable=True),
    StructField("Age", IntegerType(), nullable=True),
    StructField("Zip Code", StringType(), nullable=True),  # 强制该列为字符串
    # 补充其他列的类型定义
])

SourceFilePath = 'abfss://datalake@blobstorageaccountname.dfs.core.windows.net/SourceDir/SourceFileName.csv'
# 读取时指定自定义Schema
dfb = spark.read.load(SourceFilePath, format='csv', header=True, schema=custom_schema).cache()

DestinationPath = 'abfss://datalake@blobstorageaccountname.dfs.core.windows.net/DestinationUnknown/'
dfb.write \
    .mode("overwrite") \
    .parquet(DestinationPath)

方案2：自动生成Schema后修改目标列类型 #

如果CSV列较多，不想手动维护全量Schema，可以先让Spark自动推断Schema，再修改Zip Code列的类型，兼顾通用性和灵活性：

from pyspark.sql.types import StringType

SourceFilePath = 'abfss://datalake@blobstorageaccountname.dfs.core.windows.net/SourceDir/SourceFileName.csv'
# 先读取一次生成原始Schema
temp_df = spark.read.load(SourceFilePath, format='csv', header=True)
original_schema = temp_df.schema

# 修改Zip Code列的类型为字符串
if "Zip Code" in original_schema.names:
    modified_schema = original_schema.withField("Zip Code", StringType())
else:
    # 处理列名不存在的异常情况（可选）
    modified_schema = original_schema

# 使用修改后的Schema读取数据
dfb = spark.read.load(SourceFilePath, format='csv', header=True, schema=modified_schema).cache()

# 写入Parquet的代码保持不变
DestinationPath = 'abfss://datalake@blobstorageaccountname.dfs.core.windows.net/DestinationUnknown/'
dfb.write \
    .mode("overwrite") \
    .parquet(DestinationPath)

关键说明 #

• 必须在读取阶段指定列类型：一旦Spark自动将Zip Code推断为整数，原始数据中的前导0就会丢失，后续无法恢复。
• 两种方案都无需修改写入逻辑，保留了原有代码的通用性，仅针对读取环节做调整。

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