Azure Data Factory如何将Blob中最新文件加载至Snowflake？

适用的ADF活动 #

• 复制活动（Copy Activity）：ADF中跨数据源迁移数据的核心活动，直接支持Blob存储与Snowflake的连接，可配合动态逻辑定位目标文件。
• Get Metadata活动：用于获取Blob路径下文件的元数据（文件名、修改时间等），是实现动态筛选最新文件的基础。

最优实现方案 #

方案一：通过元数据筛选最新文件（通用场景）

1. 获取Blob路径下的文件元数据
   配置Get Metadata活动，数据源指向Blob存储的my_path路径，勾选「子项」选项，获取该路径下所有文件的完整元数据列表。

2. 筛选目标文件
   添加Filter活动，基于Get Metadata返回的结果筛选符合命名规则的最新文件：

   • 如果文件严格按当日日期命名，可直接匹配文件名：

     @startswith(item().name, 'my_file_') && equals(item().name, concat('my_file_', formatDateTime(utcnow(), 'yyyyMMdd'), '.csv'))
     

   • 如果存在文件生成延迟，可按修改时间排序取最新：

     @sort(activity('Get Metadata1').output.childItems, desc(item().lastModified))[0].name
     

3. 复制文件到Snowflake
   配置复制活动：

   • 源数据集：Blob存储，文件名设置为动态内容，引用Filter活动筛选出的文件名。
   • 目标数据集：Snowflake，指定目标表，根据CSV格式配置列分隔符（此处为|）、是否跳过首行等参数。
   • 可选：若需避免重复数据，可在Snowflake端设置截断加载逻辑，或用文件日期后缀作为增量标识实现增量加载。

方案二：直接传递生成的文件名（高效场景）

如果希望跳过元数据筛选步骤，可在原Notebook中直接输出生成的文件名，传递给后续活动：

1. 修改Notebook代码，添加输出逻辑（以Databricks Notebook为例）：

   sheetId='33222424'
   load_time = today.strftime("%Y%m%d")
   path = 'my_path'
   file_name = 'my_file_'+load_time+'.csv'
   smartsheet_client.Sheets.get_sheet_as_excel(sheetId, path, file_name)
   dbutils.notebook.exit(file_name)
   

2. 在ADF中捕获Notebook的输出值，赋值给变量。
3. 复制活动的源文件名直接引用该变量，无需额外筛选步骤，效率更高。

注意事项 #

• 确保Blob存储与Snowflake的链接服务配置正确，具备对应的读写权限。
• 确认CSV文件的列分隔符、数据格式与Snowflake目标表的结构匹配，避免解析错误。

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