基于Azure Data Factory实现跨源数据聚合统计方案咨询

前置准备 #

先完成三个数据源的链接服务与数据集配置：

• 链接服务配置
  • 本地SQL Server：使用自托管集成运行时，填入服务器地址、认证信息（SQL/Windows认证），测试确保连接正常。
  • Blob Storage：选择存储账户，用账户密钥或托管标识完成认证，指定存放dept.csv的容器，测试连接。
  • Azure SQL DB：填入服务器名、数据库名、认证信息，测试连接。
• 数据集配置
  • 本地SQL数据集：关联上述SQL Server链接服务，直接选择emp表。
  • Blob数据集：类型选DelimitedText，关联Blob链接服务，指定dept.csv路径，设置逗号为分隔符，启用"第一行作为表头"。
  • Azure DB目标数据集：关联Azure SQL链接服务，暂不指定表（后续通过活动创建或数据流自动生成）。

方案一：使用标准流水线活动实现 #

流水线活动流程如下：

1. Lookup活动1：读取员工部门数据
   配置本地SQL数据集，执行查询 SELECT dept_id, emp_id FROM emp，获取所有员工的部门关联数据。
2. Lookup活动2：读取部门名称映射
   配置Blob数据集，全量读取dept.csv的dept_id和dept-name字段（注意字段名下划线/短横线的一致性，可在数据集里重命名为dept_name避免后续报错）。
3. Join活动：关联员工与部门数据
   两个输入分别为Lookup1和Lookup2的输出，关联键设为dept_id，选择Inner Join，输出包含dept_name、emp_id的数据集。
4. Aggregate活动：统计部门员工数
   分组键选择dept_name，添加聚合规则：total_emp_count = count(emp_id)，生成按部门分组的统计结果。
5. SQL脚本活动：创建目标表（按需）
   关联Azure DB链接服务，执行脚本：

   IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'emp_details')
   CREATE TABLE emp_details (
       dept_name VARCHAR(100) NOT NULL,
       total_emp_count INT NOT NULL
   )
   

6. Copy活动：写入目标表
   源选择Aggregate活动的输出，目标选择Azure DB的emp_details数据集，设置写入行为为Overwrite或Append，完成数据写入。

方案二：使用数据流实现（更简洁） #

1. 创建数据流，添加三个数据源：
   • SQL Server源：关联本地SQL数据集，读取emp表的dept_id、emp_id字段。
   • Blob源：关联Blob数据集，读取dept_id、dept-name字段（可在源转换里将dept-name重命名为dept_name）。
   • Azure SQL目标：关联Azure DB数据集，指定目标表名emp_details，设置"如果不存在则创建表"。
2. 添加Join转换：将SQL Server源与Blob源通过dept_id做Inner Join。
3. 添加Aggregate转换：按dept_name分组，设置聚合列total_emp_count = count(emp_id)。
4. 将Aggregate转换的输出连接到Azure SQL目标，完成字段映射后即可运行数据流。

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