通过PySpark+JDBC从本地SQL Server迁移建表语句至Azure SQL

1. 从本地SQL Server提取完整CREATE TABLE语句 #

通过Databricks的JDBC连接，查询SQL Server系统视图生成包含主键、索引、排序规则的完整DDL语句：

Python示例代码 #

# 本地SQL Server JDBC连接配置
local_sql_jdbc_url = "jdbc:sqlserver://<本地SQL服务器地址>:1433;databaseName=<数据库名>;user=<用户名>;password=<密码>"

# 查询系统视图生成CREATE TABLE语句（以指定表为例）
table_name = "your_target_table"
ddl_query = f"""
SELECT 
    'CREATE TABLE [{SCHEMA_NAME(tbl.schema_id)}].[{table_name}] (' + 
    STRING_AGG(
        CONCAT(
            '[', c.name, '] ', 
            t.name, 
            CASE WHEN t.name IN ('varchar', 'nvarchar', 'char', 'nchar') THEN CONCAT('(', CASE WHEN c.max_length = -1 THEN 'MAX' ELSE CAST(c.max_length AS VARCHAR) END, ')') ELSE '' END,
            CASE WHEN c.is_nullable = 0 THEN ' NOT NULL' ELSE ' NULL' END,
            CASE WHEN c.collation_name IS NOT NULL THEN CONCAT(' COLLATE ', c.collation_name) ELSE '' END
        ), ', '
    ) + 
    ')' + 
    ISNULL((
        SELECT ', CONSTRAINT [' + k.name + '] PRIMARY KEY (' + STRING_AGG('[', c.name, ']', ', ') + ')'
        FROM sys.key_constraints k
        JOIN sys.index_columns ic ON k.parent_object_id = ic.object_id AND k.unique_index_id = ic.index_id
        JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
        WHERE k.parent_object_id = OBJECT_ID('[{SCHEMA_NAME(tbl.schema_id)}].[{table_name}]') AND k.type = 'PK'
    ), '') +
    ISNULL((
        SELECT STRING_AGG(
            CONCAT(
                ', CREATE ', CASE WHEN i.is_unique = 1 THEN 'UNIQUE ' ELSE '' END, 'INDEX [', i.name, '] ON [{SCHEMA_NAME(tbl.schema_id)}].[{table_name}] (',
                STRING_AGG(CONCAT('[', c.name, '] ', CASE WHEN ic.is_descending_key = 1 THEN 'DESC' ELSE 'ASC' END), ', '),
                ')'
            ), ' '
        )
        FROM sys.indexes i
        JOIN sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id
        JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
        WHERE i.object_id = OBJECT_ID('[{SCHEMA_NAME(tbl.schema_id)}].[{table_name}]') AND i.type <> 0 -- 排除聚集索引（主键已包含）
        GROUP BY i.name, i.is_unique
    ), '') AS create_table_ddl
FROM sys.columns c
JOIN sys.tables tbl ON c.object_id = tbl.object_id
JOIN sys.types t ON c.system_type_id = t.system_type_id AND c.user_type_id = t.user_type_id
WHERE tbl.name = '{table_name}'
GROUP BY SCHEMA_NAME(tbl.schema_id), tbl.name
"""

# 执行查询获取DDL语句
df_ddl = spark.read.jdbc(url=local_sql_jdbc_url, table=f"({ddl_query}) AS ddl_result")
create_table_stmt = df_ddl.collect()[0][0]
print(create_table_stmt)

2. 在Databricks中保存/验证表结构 #

方式1：创建内部/外部表 #

如果需要在Databricks中直接创建对应表结构，执行提取到的DDL：

# 执行DDL创建表
spark.sql(create_table_stmt)

方式2：保存DDL到DBFS #

将DDL语句保存到DBFS持久化，方便后续部署：

dbutils.fs.put("/path/to/ddl/your_table_ddl.sql", create_table_stmt, overwrite=True)

3. 部署表结构到Azure SQL数据库 #

通过Databricks的JDBC连接Azure SQL，执行提取到的CREATE TABLE语句：

Python示例代码 #

# Azure SQL数据库JDBC连接配置
azure_sql_jdbc_url = "jdbc:sqlserver://<Azure SQL服务器地址>.database.windows.net:1433;databaseName=<数据库名>;user=<用户名>;password=<密码>"

# 通过JDBC执行DDL
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
connection = spark._sc._gateway.jvm.java.sql.DriverManager.getConnection(azure_sql_jdbc_url.split(";")[0], "<Azure用户名>", "<Azure密码>")
statement = connection.createStatement()
statement.execute(create_table_stmt)
statement.close()
connection.close()

注意事项 #

• 确保本地SQL Server与Azure SQL的排序规则兼容，若存在不支持的排序规则，需手动调整DDL中的COLLATE子句。
• 索引创建需注意Azure SQL的性能限制，避免创建不必要的冗余索引。
• 若涉及多个表，可循环遍历sys.tables批量生成DDL并执行。

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