PySpark读取MSSQL表写入Snowflake时无法保留VARCHAR长度

问题根源在于Spark JDBC读取MSSQL时，会将所有VARCHAR/NVARCHAR类型转换为无长度属性的StringType，导致Snowflake Connector默认使用最大长度16777216创建列。要自动保留原始长度，需要从MSSQL元数据中提取列长度信息，再传递给Snowflake建表参数。

步骤1：从MSSQL获取列元数据 #

通过查询MSSQL系统表sys.columns和sys.types，提取目标表的列名、数据类型及长度信息：

def get_mssql_column_defs(jdbc_url, db_properties, target_table):
    # 拆分表名与schema（默认dbo）
    schema_name, table_name = (target_table.split('.', 1) if '.' in target_table else ('dbo', target_table))
    
    # 查询元数据的SQL
    meta_query = f"""
        SELECT 
            c.name AS col_name,
            t.name AS data_type,
            c.max_length
        FROM sys.columns c
        JOIN sys.types t ON c.system_type_id = t.system_type_id
        WHERE c.object_id = OBJECT_ID('{schema_name}.{table_name}')
    """
    
    # 读取元数据
    meta_df = spark.read \
        .format("jdbc") \
        .option("url", jdbc_url) \
        .option("dbtable", f"({meta_query}) AS table_meta") \
        .options(**db_properties) \
        .load()
    
    # 转换为Snowflake兼容的列定义
    snowflake_col_defs = []
    for row in meta_df.collect():
        col_name = row["col_name"]
        data_type = row["data_type"].lower()
        max_len = row["max_length"]
        
        # 处理VARCHAR/NVARCHAR类型
        if data_type in ["varchar", "nvarchar"]:
            # NVARCHAR的max_length是字节数，需转为字符数（除以2）
            actual_len = max_len // 2 if data_type == "nvarchar" else max_len
            # 处理max_length=-1（对应VARCHAR(MAX)）
            actual_len = 16777216 if actual_len <= 0 else actual_len
            snowflake_col_defs.append(f"{col_name} VARCHAR({actual_len})")
        else:
            # 其他类型直接映射（如INT→INT, DATE→DATE等）
            snowflake_col_defs.append(f"{col_name} {row['data_type'].upper()}")
    
    return ",".join(snowflake_col_defs)

步骤2：写入Snowflake时传递列定义 #

调用上述函数获取列定义，通过Snowflake Connector的create_table_column_types参数指定建表结构：

# 读取MSSQL数据（原代码不变）
df_mssql = spark.read \
    .format("jdbc") \
    .option("url", jdbcUrl) \
    .option("dbtable", "my_table") \
    .options(**mssql_properties) \
    .load()

# 获取Snowflake列定义
snowflake_col_types = get_mssql_column_defs(jdbcUrl, mssql_properties, "my_table")

# 写入Snowflake并保留原始长度
df_mssql.write \
    .format("net.snowflake.spark.snowflake") \
    .options(**snowflake_properties) \
    .option("create_table_column_types", snowflake_col_types) \
    .mode("overwrite") \
    .save()

关键注意点 #

• 权限要求：连接MSSQL的账号需要有访问sys.columns和sys.types系统表的权限。
• 特殊类型处理：代码中已处理NVARCHAR的字节数转字符数，以及VARCHAR(MAX)（max_length=-1）的情况，可根据业务需求调整。
• 类型映射：其他数据类型（如INT、DATE、DECIMAL等）会自动映射为Snowflake兼容类型，若有特殊映射需求可在函数中扩展。

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