Azure 数据工厂和 Azure Synapse Analytics 中的查找活动
适用于：
Azure 数据工厂 Azure Synapse Analytics
提示
试用 Microsoft Fabric 中的数据工厂，这是一种适用于企业的一站式分析解决方案。 Microsoft Fabric 涵盖从数据移动到数据科学、实时分析、商业智能和报告的所有内容。 了解如何免费开始新的试用！
查找活动可以从数据工厂和 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 管道支持的其他控制流活动：