使用 Azure 数据工厂将源文件中的列与接收器 table 列匹配以确保它们匹配
matching the columns in a source file with sink table columns to make sure they match using Azure Data Factory
我有一个 Azure 数据工厂触发器，当文件被放置在 blob 存储中时，该触发器将被触发，该触发器将开始管道执行并将文件名传递给数据流 activity。我想确保文件中 header 行的所有列名都在接收器 table 中。接收器 table 中有一个不应该在比较中的标识列。不知道如何解决这个任务，我读过 'derived column' activity，这是我应该走的路吗？
策略：
使用两个 ADF 管道，一个获取所有文件的列表，另一个处理每个文件，将其内容复制到特定 SQL table.
设置：
我按照您需要的模式创建了 4 个 CSV 文件：“[CustomerID][TableName][FileID].csv”和 4 SQL tables，每种文件一个。
- A_inventory_0001.csv：客户A的库存记录，待
插入 SQL table “A_Inventory”.
- A_sales_0003.csv：销售额
客户 A 的记录，将被插入 SQL table “A_Sales”.
- B_inventory_0002.csv：客户B的库存记录，待
插入 SQL table “B_Inventory”.
- B_sales_0004.csv：销售额
客户 B 的记录，将被插入 SQL table “B_Sales”
链接服务
在 Azure 数据工厂中，以下链接服务是使用 Key Vault 创建的（Key Vault 是可选的）。
数据集
创建了以下数据集。请注意，我们已经创建了一些参数以允许管道指定源文件和目标 SQL table.
数据集“AzureSQLTable”有一个指定目的地名称的参数SQLtable.
数据集“DelimitedTextFile”有一个参数来指定源 CSV 文件的名称。
数据集“DelimitedTextFiles”没有参数，因为它将用于列出源文件夹中的所有文件。
管道
第一个管道“获取文件”将从源文件夹中获取 CSV 文件列表（获取元数据 activity），然后，对于每个文件，调用第二个管道将 CSV 文件名作为参数.
在 foreach 循环中，调用第二个管道“Process File”，将文件名作为参数传递。
第二个管道有一个参数“pFileName”来接收要处理的文件的名称和一个根据文件名计算目标名称的变量table。
第一个activity是在文件名中使用拆分来提取我们需要组成目标table名称的部分。
在下面的表达式中，我们使用“__”分隔符拆分文件名，然后使用第一部分和第二部分组成目标 table 名称。
@concat(string(split(pipeline().parameters.pFileName, '_')[0]),'_',string(split(pipeline().parameters.pFileName, '_')[10]))
然后第二个activity将使用动态映射将文件从源“pFileName”复制到目的地table“vTableName”，即不添加特定的列名因为这将是动态的。
我在此示例中使用的文件和 ADF 代码可在此处获得：
https://github.com/diegoeick/stack-overflow/tree/main/69340699
希望这能解决您的问题。
如果您仍然需要将 CustomerID 和 FileID 保存在数据库 tables 中，您可以使用动态映射并使用可用参数（文件名）并创建一个 json在您的副本 activity 的映射选项卡中进行动态映射。您可以在此处找到更多详细信息：https://docs.microsoft.com/en-us/azure/data-factory/copy-activity-schema-and-type-mapping#parameterize-mapping
您可以 select 或使用“字段映射”筛选驻留在接收器数据集中的列或 table。您可以选择使用“derived columns”转换，但是在“sink transformation”中您将默认使用此转换并设置为“自动映射”。您可以在此处添加或删除写入接收器的列。
在下面的示例中，列“id”可以假定为类似于您的 table 中的“Identity”列。假设所有文件都有相同的列：
根据需要修改后，您可以在 运行 之前的“inspect”选项卡中进行确认。
我有一个 Azure 数据工厂触发器，当文件被放置在 blob 存储中时，该触发器将被触发，该触发器将开始管道执行并将文件名传递给数据流 activity。我想确保文件中 header 行的所有列名都在接收器 table 中。接收器 table 中有一个不应该在比较中的标识列。不知道如何解决这个任务，我读过 'derived column' activity，这是我应该走的路吗？
策略：
使用两个 ADF 管道，一个获取所有文件的列表，另一个处理每个文件，将其内容复制到特定 SQL table.
设置：
我按照您需要的模式创建了 4 个 CSV 文件：“[CustomerID][TableName][FileID].csv”和 4 SQL tables，每种文件一个。
- A_inventory_0001.csv：客户A的库存记录，待 插入 SQL table “A_Inventory”.
- A_sales_0003.csv：销售额 客户 A 的记录，将被插入 SQL table “A_Sales”.
- B_inventory_0002.csv：客户B的库存记录，待 插入 SQL table “B_Inventory”.
- B_sales_0004.csv：销售额 客户 B 的记录，将被插入 SQL table “B_Sales”
链接服务
在 Azure 数据工厂中，以下链接服务是使用 Key Vault 创建的（Key Vault 是可选的）。
数据集
创建了以下数据集。请注意，我们已经创建了一些参数以允许管道指定源文件和目标 SQL table.
数据集“AzureSQLTable”有一个指定目的地名称的参数SQLtable.
数据集“DelimitedTextFile”有一个参数来指定源 CSV 文件的名称。
数据集“DelimitedTextFiles”没有参数，因为它将用于列出源文件夹中的所有文件。
管道
第一个管道“获取文件”将从源文件夹中获取 CSV 文件列表（获取元数据 activity），然后，对于每个文件，调用第二个管道将 CSV 文件名作为参数.
在 foreach 循环中，调用第二个管道“Process File”，将文件名作为参数传递。
第二个管道有一个参数“pFileName”来接收要处理的文件的名称和一个根据文件名计算目标名称的变量table。
第一个activity是在文件名中使用拆分来提取我们需要组成目标table名称的部分。
在下面的表达式中，我们使用“__”分隔符拆分文件名，然后使用第一部分和第二部分组成目标 table 名称。
@concat(string(split(pipeline().parameters.pFileName, '_')[0]),'_',string(split(pipeline().parameters.pFileName, '_')[10]))
然后第二个activity将使用动态映射将文件从源“pFileName”复制到目的地table“vTableName”，即不添加特定的列名因为这将是动态的。
我在此示例中使用的文件和 ADF 代码可在此处获得： https://github.com/diegoeick/stack-overflow/tree/main/69340699
希望这能解决您的问题。
如果您仍然需要将 CustomerID 和 FileID 保存在数据库 tables 中，您可以使用动态映射并使用可用参数（文件名）并创建一个 json在您的副本 activity 的映射选项卡中进行动态映射。您可以在此处找到更多详细信息：https://docs.microsoft.com/en-us/azure/data-factory/copy-activity-schema-and-type-mapping#parameterize-mapping
您可以 select 或使用“字段映射”筛选驻留在接收器数据集中的列或 table。您可以选择使用“derived columns”转换，但是在“sink transformation”中您将默认使用此转换并设置为“自动映射”。您可以在此处添加或删除写入接收器的列。
在下面的示例中，列“id”可以假定为类似于您的 table 中的“Identity”列。假设所有文件都有相同的列：
根据需要修改后，您可以在 运行 之前的“inspect”选项卡中进行确认。