如何在Azure Data Factory中从CSV文件自动创建Azure SQL DB表？单文件单表实现

核心思路 #

通过Get Metadata活动获取CSV文件列表，配合For Each循环遍历每个文件，在循环内实现「自动建表+数据导入」的逻辑。以下是两种常用实现方案：

方案一：用Copy Activity自动建表（快速实现） #

这种方法利用ADF Copy Activity的「自动创建表」功能，无需手动编写SQL，适合快速落地。

步骤1：配置Get Metadata活动获取CSV列表 #

• 新建Get Metadata活动（命名示例：Get_CSV_Files）
• 链接服务：选择CSV所在的存储服务（如Blob Storage、ADLS Gen2）
• 数据集：指向存储中的CSV文件夹（注意是文件夹，而非单个文件）
• 字段列表：勾选Child Items，活动输出会包含文件夹下所有文件的名称、路径等信息

步骤2：配置For Each循环遍历文件 #

• 新建For Each活动（命名示例：Loop_Through_CSVs）
• 迭代项：填入动态内容 @activity('Get_CSV_Files').output.childItems
• 可选：勾选「Sequential」（顺序执行），避免并发操作导致的冲突

步骤3：在循环内配置Copy Activity #

在For Each活动内部添加Copy Activity，完成单文件的建表与数据导入：

源端配置

• 链接服务：同Get Metadata使用的存储服务
• 数据集：选择CSV数据集，在「文件名」字段填入动态内容 @item().name（每次循环指向当前CSV文件）
• 格式设置：勾选「第一行为标题」，确保ADF识别列名

目标端配置

• 链接服务：选择Azure SQL DB的链接服务
• 数据集：选择SQL DB数据集，在「表名」字段填入动态内容 @replace(item().name, '.csv', '')（自动将CSV文件名转为表名，去掉.csv后缀）
• 勾选「Auto create table」选项，ADF会自动根据CSV的字段类型推断SQL表结构并创建表
• 映射：保持默认自动映射即可（源CSV的列名会与目标表列名自动匹配）

方案二：用Lookup+Script活动自定义建表（灵活可控） #

如果需要自定义数据类型、处理特殊列名，可先通过Lookup获取CSV schema，再用Script活动执行自定义建表语句。

步骤1-2：同方案一的Get Metadata+For Each循环 #

步骤3：在循环内添加Lookup活动获取CSV schema #

• 新建Lookup活动（命名示例：Get_CSV_Schema）
• 源配置：同方案一的CSV源，指向当前循环的文件 @item().name
• 设置：勾选「第一行为标题」，并将「行数」设为1（仅读取第一行数据用于推断类型）

步骤4：在循环内添加Script活动执行建表语句 #

• 新建Script活动（命名示例：Create_SQL_Table）
• 链接服务：选择Azure SQL DB的链接服务
• 脚本内容：用动态内容拼接自定义建表SQL，示例如下（可根据需求调整数据类型逻辑）：

CREATE TABLE IF NOT EXISTS @{replace(replace(item().name, '.csv', ''), ' ', '_')} (
@{substring(
    join(
        map(
            items(activity('Get_CSV_Schema').output.firstRow),
            concat('    ', item().key, ' ', 
                if(contains(string(item().value), '.'), 'FLOAT', 
                    if(isInteger(item().value), 'INT', 
                        if(isDate(item().value), 'DATE', 'VARCHAR(MAX)')
                    )
                ), ','
            )
        ),
        '\n'
    ),
    0,
    sub(
        length(
            join(
                map(
                    items(activity('Get_CSV_Schema').output.firstRow),
                    concat('    ', item().key, ' ', 
                        if(contains(string(item().value), '.'), 'FLOAT', 
                            if(isInteger(item().value), 'INT', 
                                if(isDate(item().value), 'DATE', 'VARCHAR(MAX)')
                            )
                        ), ','
                    )
                ),
                '\n'
            )
        ),
        1
    )
)}
)

• 说明：脚本会自动处理文件名中的空格（替换为下划线），根据CSV第一行数据推断SQL数据类型，若类型推断不符合需求，可修改if逻辑调整。

步骤5：添加Copy Activity导入数据 #

同方案一的Copy Activity配置，将当前CSV的数据导入已创建的SQL表中。

关键注意事项 #

• 文件名清洗：若CSV文件名包含SQL不允许的字符（如空格、-、#等），需用动态内容处理，例如 @replace(replace(item().name, '.csv', ''), ' ', '_') 将空格替换为下划线
• 数据类型调整：ADF自动推断的类型可能存在偏差（如长文本被设为VARCHAR(255)），可通过修改CSV数据集的字段类型，或自定义Script活动的SQL语句优化
• 重复执行处理：若需要覆盖已有表，可在Copy Activity前添加Script活动执行 DROP TABLE IF EXISTS [表名]

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