在ADF的For Each循环中批量转换Timestamp类型为字符串并加载多表至ADLS

方法一：使用ADF数据流动（Data Flow）实现批量类型转换 #

这是最直观且易维护的方案，适合批量处理表的类型转换需求：

• 在For Each循环内替换Copy Data活动为数据流动活动，将循环项的表名作为参数传入数据流动
• 数据流动内配置SQL Server数据源，通过参数指定要读取的表（表达式示例：@pipeline().parameters.tableName）
• 添加派生列转换，切换到高级模式，使用以下表达式批量处理所有列：

  mapColumns(columns(), 
      c => if(typeOf(c) == 'timestamp', 
          toString(c, 'yyyy-MM-dd HH:mm:ss.fff'), 
          c
      )
  )
  

  该表达式会遍历所有列，自动识别timestamp类型列并转换为指定格式的字符串，其他列保持原有类型和值
• 配置ADLS数据目标，根据需求选择存储格式（如Parquet、CSV），并参数化文件路径/名称

方法二：动态生成Copy Data活动的映射规则 #

如果坚持使用Copy Data活动，可以通过动态生成映射JSON实现自动转换：

1. 获取表列结构：在For Each循环内添加Lookup活动，执行SQL查询获取当前表的列信息：

   SELECT COLUMN_NAME, DATA_TYPE 
   FROM INFORMATION_SCHEMA.COLUMNS 
   WHERE TABLE_NAME = '@item().tableName' 
     AND TABLE_SCHEMA = 'dbo' -- 根据实际架构调整
   

2. 生成动态映射：添加Set Variable活动，创建字符串类型变量，用以下表达式生成符合ADF映射格式的JSON：

   @concat(
       '{ "translator": { "type": "TabularTranslator", "mappings": [',
       join(
           foreach(activity('Lookup列信息').output.value,
               concat(
                   '{ "source": { "name": "', item().COLUMN_NAME, '" }, ',
                   '"sink": { "name": "', item().COLUMN_NAME, '" }, ',
                   '"typeConversion": true, ',
                   '"typeConversionSettings": { ',
                       '"sourceDataType": "', if(item().DATA_TYPE == 'timestamp', 'timestamp', ''), '", ',
                       '"targetDataType": "', if(item().DATA_TYPE == 'timestamp', 'string', item().DATA_TYPE), '" ',
                   '} }'
               )
           ),
           ','
       ),
       '] } }'
   )
   

3. 应用动态映射：在Copy Data活动的「映射」选项卡，选择「使用动态内容」，将上述变量填入即可

注意事项 #

• 使用Data Flow时，确认SQL Server数据源的类型识别正确，timestamp类型在Data Flow中对应的数据类型为timestamp（需与datetime/datetime2区分）
• 动态映射方案中，需根据ADLS目标格式调整非timestamp列的类型映射，避免类型不兼容问题
• 若表数量较多，可开启For Each循环的并行执行，但需控制并发数，避免超出SQL Server的连接限制

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