Azure 数据工厂和 Azure Synapse Analytics 中的查找活动
适用于:Azure 数据工厂 Azure Synapse Analytics
查找活动可以从数据工厂和 Synapse 管道支持的任何数据源中检索数据集。 可以使用该活动来动态确定哪些对象在后续活动中工作,而不是针对对象名称进行硬编码。 一些对象示例包括文件和表。
查找活动读取并返回配置文件或表的内容。 它还返回执行查询或存储过程的结果。 输出可以是单一实例值或属性数组,可在后续的复制、转换或控制流活动(如 ForEach 活动)中使用它们。
通过 UI 创建 Lookup 活动
若要在管道中使用 Lookup 活动,请完成以下步骤:
在管道“活动”窗格中搜索“Lookup”,然后将 Lookup 活动拖动到管道画布上。
如果尚未选择画布上的新 Lookup 活动,请选择它及其“设置”选项卡,以编辑其详细信息。
选择现有源数据集,或选择“新建”按钮创建新的源数据集。
用于标识源数据集中要包括的行的选项因数据集类型而异。 上面的示例显示了带分隔符的文本数据集的配置选项。 下面是 Azure SQL 表数据集和 OData 数据集的配置选项示例。
支持的功能
注意以下事项:
- 查找活动最多可以返回 5000 行;如果结果集包含的记录超过此范围,将返回前 5000 行。
- 查找活动的输出最大支持 4 MB,如果大小超过此限制,活动将失败。
- 查找活动在超时前的最长持续时间为 24 小时。
注意
当使用查询或存储过程查找数据时,请确保返回一个结果集,并且只返回一个。 否则,查找活动将失败。
查找活动支持以下数据源。
注意
连接器标记为“预览”意味着,可以试用它并向我们提供反馈。 若要在解决方案中使用预览版连接器的依赖项,请联系 Azure 客户支持。
语法
{
"name":"LookupActivity",
"type":"Lookup",
"typeProperties":{
"source":{
"type":"<source type>"
},
"dataset":{
"referenceName":"<source dataset name>",
"type":"DatasetReference"
},
"firstRowOnly":<true or false>
}
}
Type 属性
名称 | 说明 | 类型 | 必需? |
---|---|---|---|
dataset | 为查找提供数据集引用。 从每篇相应的连接器文章的“数据集属性”部分中获取详细信息。 | 键/值对 | 是 |
source | 包含特定于数据集的源属性,与复制活动源相同。 从每篇相应的连接器文章的“复制活动属性”部分中获取详细信息。 | 键/值对 | 是 |
firstRowOnly | 指示仅返回第一行还是返回所有行。 | Boolean | 否。 默认为 true 。 |
注意
- 不支持 ByteArray 类型的源列。
- 数据集定义不支持结构。 对于文本格式化文件,使用标头行提供列名。
- 如果查找源是 JSON 文件,则不支持用于重塑 JSON 对象的
jsonPathDefinition
设置。 将检索整个对象。
使用查找活动结果
查找结果会返回到活动运行结果的 output
节。
当
firstRowOnly
设置为true
(默认值)时,输出格式如以下代码所示。 查找结果位于固定的firstRow
键下。 若要在后续活动中使用该结果,请使用@{activity('LookupActivity').output.firstRow.table}
模式。{ "firstRow": { "Id": "1", "schema":"dbo", "table":"Table1" } }
当
firstRowOnly
设置为false
时,输出格式如以下代码所示。count
字段指示返回的记录数。 详细值显示在固定的value
数组下。 在这种情况下,查找活动后跟 Foreach 活动。 使用@activity('MyLookupActivity').output.value
模式将value
数组传递给 ForEach 活动items
字段。 若要访问value
数组中的元素,请使用以下语法:@{activity('lookupActivity').output.value[zero based index].propertyname}
。 示例为@{activity('lookupActivity').output.value[0].schema}
。{ "count": "2", "value": [ { "Id": "1", "schema":"dbo", "table":"Table1" }, { "Id": "2", "schema":"dbo", "table":"Table2" } ] }
示例
在此示例中,管道包含两个活动:查找和复制 。 复制活动将 Azure SQL 数据库实例的 SQL 表中的数据复制到 Azure Blob 存储。 SQL 表的名称存储在 Blob 存储的 JSON 文件中。 查找活动在运行时查找表名。 使用此方法动态修改 JSON。 不需要重新部署管道或数据集。
本示例演示如何只查找第一行。 若要查找所有行并将结果与 ForEach 活动链接,请参阅批量复制多个表中的示例。
管道
- 查找活动配置为使用 LookupDataset,该项引用 Azure Blob 存储中的一个位置。 查找活动在此位置从 JSON 文件读取 SQL 表名称。
- 复制活动使用查找活动的输出,即 SQL 表的名称。 SourceDataset 中的 tableName 属性配置为使用查找活动的输出。 复制活动将数据从 SQL 表复制到 Azure Blob 存储中的一个位置。 该位置由 SinkDataset 属性指定。
{
"name": "LookupPipelineDemo",
"properties": {
"activities": [
{
"name": "LookupActivity",
"type": "Lookup",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "JsonSource",
"storeSettings": {
"type": "AzureBlobStorageReadSettings",
"recursive": true
},
"formatSettings": {
"type": "JsonReadSettings"
}
},
"dataset": {
"referenceName": "LookupDataset",
"type": "DatasetReference"
},
"firstRowOnly": true
}
},
{
"name": "CopyActivity",
"type": "Copy",
"dependsOn": [
{
"activity": "LookupActivity",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "AzureSqlSource",
"sqlReaderQuery": {
"value": "select * from [@{activity('LookupActivity').output.firstRow.schema}].[@{activity('LookupActivity').output.firstRow.table}]",
"type": "Expression"
},
"queryTimeout": "02:00:00",
"partitionOption": "None"
},
"sink": {
"type": "DelimitedTextSink",
"storeSettings": {
"type": "AzureBlobStorageWriteSettings"
},
"formatSettings": {
"type": "DelimitedTextWriteSettings",
"quoteAllText": true,
"fileExtension": ".txt"
}
},
"enableStaging": false,
"translator": {
"type": "TabularTranslator",
"typeConversion": true,
"typeConversionSettings": {
"allowDataTruncation": true,
"treatBooleanAsNumber": false
}
}
},
"inputs": [
{
"referenceName": "SourceDataset",
"type": "DatasetReference",
"parameters": {
"schemaName": {
"value": "@activity('LookupActivity').output.firstRow.schema",
"type": "Expression"
},
"tableName": {
"value": "@activity('LookupActivity').output.firstRow.table",
"type": "Expression"
}
}
}
],
"outputs": [
{
"referenceName": "SinkDataset",
"type": "DatasetReference",
"parameters": {
"schema": {
"value": "@activity('LookupActivity').output.firstRow.schema",
"type": "Expression"
},
"table": {
"value": "@activity('LookupActivity').output.firstRow.table",
"type": "Expression"
}
}
}
]
}
],
"annotations": [],
"lastPublishTime": "2020-08-17T10:48:25Z"
}
}
查找数据集
查找数据集是指由 AzureBlobStorageLinkedService 类型指定的 Azure 存储查找文件夹中的 sourcetable.json 文件 。
{
"name": "LookupDataset",
"properties": {
"linkedServiceName": {
"referenceName": "AzureBlobStorageLinkedService",
"type": "LinkedServiceReference"
},
"annotations": [],
"type": "Json",
"typeProperties": {
"location": {
"type": "AzureBlobStorageLocation",
"fileName": "sourcetable.json",
"container": "lookup"
}
}
}
}
复制活动的源数据集
源数据集使用查找活动的输出,即 SQL 表名称。 复制活动将数据从此 SQL 表复制到 Azure Blob 存储中的一个位置。 该位置由接收器数据集指定。
{
"name": "SourceDataset",
"properties": {
"linkedServiceName": {
"referenceName": "AzureSqlDatabase",
"type": "LinkedServiceReference"
},
"parameters": {
"schemaName": {
"type": "string"
},
"tableName": {
"type": "string"
}
},
"annotations": [],
"type": "AzureSqlTable",
"schema": [],
"typeProperties": {
"schema": {
"value": "@dataset().schemaName",
"type": "Expression"
},
"table": {
"value": "@dataset().tableName",
"type": "Expression"
}
}
}
}
复制活动的接收器数据集
复制活动将数据从 SQL 表复制到 Azure 存储中 csv 文件夹下的 filebylookup.csv 文件。 该文件由 AzureBlobStorageLinkedService 属性指定。
{
"name": "SinkDataset",
"properties": {
"linkedServiceName": {
"referenceName": "AzureBlobStorageLinkedService",
"type": "LinkedServiceReference"
},
"parameters": {
"schema": {
"type": "string"
},
"table": {
"type": "string"
}
},
"annotations": [],
"type": "DelimitedText",
"typeProperties": {
"location": {
"type": "AzureBlobStorageLocation",
"fileName": {
"value": "@{dataset().schema}_@{dataset().table}.csv",
"type": "Expression"
},
"container": "csv"
},
"columnDelimiter": ",",
"escapeChar": "\\",
"quoteChar": "\""
},
"schema": []
}
}
sourcetable.json
可将以下两种格式用于 sourcetable.json 文件。
对象集
{
"Id":"1",
"schema":"dbo",
"table":"Table1"
}
{
"Id":"2",
"schema":"dbo",
"table":"Table2"
}
对象数组
[
{
"Id": "1",
"schema":"dbo",
"table":"Table1"
},
{
"Id": "2",
"schema":"dbo",
"table":"Table2"
}
]
限制和解决方法
以下是 Lookup 活动的一些限制以及建议的解决方法。
限制 | 解决方法 |
---|---|
查找活动最多有 5,000 行,最大大小为 4 MB。 | 设计一个外部管道对内部管道进行迭代的两级管道,该管道会检索不超过最大行数或大小的数据。 |
相关内容
查看 Azure 数据工厂和 Synapse 管道支持的其他控制流活动: