Azure Data Factory中如何对比CSV与数据库Schema后再写入数据？

方案一：Pipeline原生活动实现（适合需灵活控制分支逻辑的场景） #

步骤1：获取数据库表Schema #

使用Lookup活动连接目标数据库，执行系统查询获取表的列名及顺序（精确匹配需包含列顺序）：

-- 以SQL Server/Azure SQL为例，其他数据库可替换对应系统表查询
SELECT COLUMN_NAME, ORDINAL_POSITION 
FROM INFORMATION_SCHEMA.COLUMNS 
WHERE TABLE_NAME = '<目标表名>' 
  AND TABLE_SCHEMA = '<目标Schema名>' 
ORDER BY ORDINAL_POSITION

该查询返回按实际存储顺序排列的列名集合。

步骤2：获取CSV文件表头 #

使用Get Metadata活动指向Blob Storage中的目标CSV文件，勾选Column Names和Column Count选项，活动输出会包含CSV的表头数组。

步骤3：转换数据并对比 #

1. 创建两个字符串数组类型的变量：DB_Columns和CSV_Columns
2. 用Set Variable活动给DB_Columns赋值，提取Lookup结果中的列名并保持顺序：

   @select(orderBy(activity('Lookup_DBSchema').output.value, asc(item().ORDINAL_POSITION)), 'COLUMN_NAME')
   

3. 用Set Variable活动给CSV_Columns赋值，直接引用Get Metadata的输出：

   @activity('GetMetadata_CSV').output.columnNames
   

4. 使用If Condition活动，判断条件为：

   @equals(variables('DB_Columns'), variables('CSV_Columns'))
   

步骤4：分支处理 #

• 条件满足（匹配）：继续执行后续的数据库写入活动
• 条件不满足（不匹配）：抛出自定义错误（用Fail活动）或触发告警逻辑，终止写入流程

方案二：Dataflow实现（适合复杂Schema自动匹配场景） #

步骤1：配置数据源 #

• 连接Blob Storage的CSV源：开启Allow Schema Drift，确保能读取动态表头
• 连接目标数据库表作为Sink：同样开启Allow Schema Drift

步骤2：添加Assert转换做Schema校验 #

在CSV源之后添加Assert转换，配置两个校验规则：

1. 列名完全匹配：

   names() == names(sink())
   

2. 列顺序完全匹配：

   indices() == indices(sink())
   

将Assert的Failure Action设置为Fail Pipeline，一旦校验不通过直接终止流程。

步骤3：执行数据写入 #

Assert校验通过后，通过Sink活动将CSV数据写入数据库，开启Auto Mapping即可自动匹配列（因Schema已校验一致）

关键注意事项 #

• 若数据库Schema频繁变更，Lookup活动的查询会自动获取最新Schema，无需硬编码
• 精确匹配需同时校验列名和顺序，避免仅列名相同但顺序不一致导致的数据写入错误
• 若CSV存在多余列或缺失列，两种方案都会触发不匹配的分支逻辑

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