从本地Teradata迁移至Snowflake时DDL脚本自动生成方法咨询

整套方案完全基于Teradata和Snowflake原生能力、通用脚本语言实现，无需引入Roboquery等第三方付费工具，可支持万级表量的批量DDL生成。

步骤1：从Teradata侧批量拉取最新元数据 #

直接调用Teradata原生DBC系统库获取全量对象的元数据，无需额外权限开通：

• 核心查询系统表包括：DBC.TablesV（库、Schema、表/视图基础属性）、DBC.ColumnsV（字段名、类型、长度、非空约束）、DBC.IndicesV（主键、索引规则）、DBC.TableConstraintsV（外键、唯一约束等）、DBC.TablesV.RequestText（视图、存储过程的原始创建语句）
• 可根据业务需求筛选需要迁移的库、对象范围，查询结果直接导出结构化数据存储，示例元数据查询SQL如下：

-- Teradata侧拉取表字段元数据并做初步类型映射示例
SELECT 
  DatabaseName,
  TableName,
  ColumnName,
  ColumnType,
  ColumnLength,
  Nullable,
  DecimalTotalDigits,
  DecimalFractionalDigits
FROM DBC.ColumnsV 
WHERE DatabaseName IN ('待迁移业务库1','待迁移业务库2')
ORDER BY DatabaseName, TableName, ColumnId;

步骤2：按技术栈选择DDL自动拼接逻辑 #

根据团队技术栈选择以下两种无第三方工具的拼接方案，均可实现实时触发、秒级生成全量DDL：

方案A：SQL直接拼接（适合无代码开发能力的场景） #

基于第一步拉取的元数据，在任意支持SQL的环境（Teradata、本地MySQL、甚至Excel函数）直接拼接生成标准Snowflake DDL语句：

-- 拼接建表语句示例
SELECT 
'CREATE OR REPLACE TABLE '||DatabaseName||'.'||TableName||' ('||
LISTAGG(
  ColumnName||' '||
  CASE 
    WHEN ColumnType IN ('I','I2','I8') THEN 'INTEGER'
    WHEN ColumnType = 'D' THEN 'DECIMAL('||DecimalTotalDigits||','||DecimalFractionalDigits||')'
    WHEN ColumnType = 'CV' THEN 'VARCHAR('||ColumnLength||')'
    WHEN ColumnType = 'TS' THEN 'TIMESTAMP_NTZ'
    ELSE 'STRING' 
  END ||
  CASE WHEN Nullable = 'N' THEN ' NOT NULL' ELSE '' END
, ', ')
||');' AS CREATE_TABLE_DDL
FROM 元数据表
GROUP BY DatabaseName, TableName;

视图、存储过程类对象直接从RequestText字段拉取原始语句，批量替换Teradata专属语法（如QUALIFY、Teradata特有系统函数）为Snowflake兼容写法即可。

方案B：Python/Shell脚本生成（适合有定制化需求、需要全流程自动化的场景） #

用通用脚本语言连接Teradata拉取元数据，自动转换后可直接提交到Snowflake执行，全程无需人工介入：

# Teradata到Snowflake类型映射字典示例
TYPE_MAPPING = {
    'I': 'INT',
    'I2': 'SMALLINT',
    'I8': 'BIGINT',
    'D': 'DECIMAL',
    'CV': 'VARCHAR',
    'TS': 'TIMESTAMP_NTZ',
    'DA': 'DATE'
}

# 批量生成DDL核心逻辑
import teradatasql
import snowflake.connector

# 连接Teradata拉取元数据
with teradatasql.connect(host='td_host', user='td_user', password='td_pwd') as td_conn:
    with td_conn.cursor() as td_cur:
        td_cur.execute("元数据查询SQL")
        meta_data = td_cur.fetchall()

# 按库表分组后拼接DDL
ddl_list = []
current_db = None
current_table = None
cols = []
for row in meta_data:
    db, table, col_name, col_type, col_len, nullable, total_digits, frac_digits = row
    if (db, table) != (current_db, current_table):
        if current_db is not None:
            ddl = f"CREATE OR REPLACE TABLE {current_db}.{current_table} (\n" + ",\n".join(cols) + "\n);"
            ddl_list.append(ddl)
        current_db, current_table = db, table
        cols = []
    # 拼接字段定义
    col_type_str = TYPE_MAPPING.get(col_type, 'STRING')
    if col_type == 'D':
        col_type_str = f"DECIMAL({total_digits},{frac_digits})"
    elif col_type == 'CV':
        col_type_str = f"VARCHAR({col_len})"
    col_def = f"{col_name} {col_type_str}" + (" NOT NULL" if nullable == 'N' else "")
    cols.append(col_def)

# 直接提交到Snowflake执行
with snowflake.connector.connect(account='sf_account', user='sf_user', password='sf_pwd') as sf_conn:
    with sf_conn.cursor() as sf_cur:
        for ddl in ddl_list:
            sf_cur.execute(ddl)

将上述逻辑封装成接口或者定时任务，即可实现DDL的实时自动生成与执行。

步骤3：特殊对象补全与校验 #

• 分区表、聚簇表等特殊表属性，单独从Teradata系统表拉取分区规则，转换为Snowflake的聚类键、分区规则补充到DDL中
• 生成完所有DDL后，可批量查询Snowflake的INFORMATION_SCHEMA校验结构一致性，避免类型映射错误
• 存储过程、自定义函数等数量较少的特殊对象，可基于元数据批量替换语法后单独生成

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