Azure 数据工厂和 Azure Synapse Analytics 中的查找活动

适用于： Azure 数据工厂 Azure Synapse Analytics

查找活动可以从数据工厂和 Synapse 管道支持的任何数据源中检索数据集。 可以使用该活动来动态确定哪些对象在后续活动中工作，而不是针对对象名称进行硬编码。 一些对象示例包括文件和表。

查找活动读取并返回配置文件或表的内容。 它还返回执行查询或存储过程的结果。 输出可以是单一实例值或属性数组，可在后续的复制、转换或控制流活动（如 ForEach 活动）中使用它们。

通过 UI 创建 Lookup 活动

若要在管道中使用 Lookup 活动，请完成以下步骤：

1. 在管道“活动”窗格中搜索“Lookup”，然后将 Lookup 活动拖动到管道画布上。

2. 如果尚未选择画布上的新 Lookup 活动，请选择它及其“设置”选项卡，以编辑其详细信息。

   

3. 选择现有源数据集，或选择“新建”按钮创建新的源数据集。

4. 用于标识源数据集中要包括的行的选项因数据集类型而异。 上面的示例显示了带分隔符的文本数据集的配置选项。 下面是 Azure SQL 表数据集和 OData 数据集的配置选项示例。

   

   

支持的功能

注意以下事项：

• 查找活动最多可以返回 5000 行；如果结果集包含的记录超过此范围，将返回前 5000 行。
• 查找活动的输出最大支持 4 MB，如果大小超过此限制，活动将失败。
• 查找活动在超时前的最长持续时间为 24 小时。

查找活动支持以下数据源。

语法

{
    "name":"LookupActivity",
    "type":"Lookup",
    "typeProperties":{
        "source":{
            "type":"<source type>"
        },
        "dataset":{
            "referenceName":"<source dataset name>",
            "type":"DatasetReference"
        },
        "firstRowOnly":<true or false>
    }
}

Type 属性

使用查找活动结果

查找结果会返回到活动运行结果的 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 活动的一些限制以及建议的解决方法。

相关内容

查看 Azure 数据工厂和 Synapse 管道支持的其他控制流活动：

• 执行管道活动
• ForEach 活动
• GetMetadata 活动
• Web 活动