Azure 数据工厂中的查找活动

适用于: Azure 数据工厂

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

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

支持的功能

注意以下事项:

  • 查找活动最多可以返回 5000 行;如果结果集包含的记录超过此范围,将返回前 5000 行。
  • 查找活动的输出最大支持 4 MB,如果大小超过此限制,活动将失败。
  • 查找活动在超时前的最长持续时间为 24 小时。
  • 当使用查询或存储过程查找数据时,请确保返回一个结果集,并且只返回一个。 否则,查找活动将失败。

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

Category 数据存储
Azure Azure Blob 存储
  Azure Cosmos DB (SQL API)
  Azure 数据资源管理器(Kusto)
  Azure Data Lake Storage Gen2
  Azure Database for MariaDB
  Azure Database for MySQL
  Azure Database for PostgreSQL
  Azure Databricks Delta Lake
  Azure 文件
  Azure SQL 数据库
  Azure SQL 数据库托管实例
  Azure Synapse Analytics
  Azure 表存储
Database Amazon Redshift
  Apache Impala
  Azure SQL 数据库托管实例
  DB2
  Drill
  Google AdWords
  Google BigQuery
  Greenplum
  HBase
  Hive
  Informix
  MariaDB
  Microsoft Access
  MySQL
  Netezza
  Oracle
  Phoenix
  PostgreSQL
  Presto
  SAP Business Warehouse Open Hub
  通过 MDX 实现的 SAP Business Warehouse
  SAP HANA
  SAP Table
  SQL Server
  Spark
  Sybase
  Teradata
  Vertica
NoSQL Cassandra
  Couchbase(预览)
文件 Amazon S3
  Amazon S3 兼容存储
  文件系统
  FTP
  Google Cloud Storage
  HDFS
  泛型 HTTP
  Oracle 云存储
  SFTP
通用协议 泛型 OData
  泛型 ODBC
  REST
  SharePoint Online 列表
服务和应用 Amazon Marketplace Web Service
  Concur(预览)
  Dataverse
  Dynamics 365
  Dynamics AX
  Dynamics CRM
  HubSpot
  Jira
  Azure 数据资源管理器(Kusto)
  Magento(预览)
  Marketo(预览)
  Office 365
  Oracle Eloqua(预览)
  Oracle Responsys(预览)
  Oracle 服务云(预览)
  PayPal(预览)
  QuickBooks(预览)
  REST
  SAP Cloud for Customer (C4C)
  SAP ECC
  Salesforce
  Salesforce Marketing Cloud
  Salesforce 服务云
  ServiceNow
  Shopify(预览)
  Snowflake
  Square(预览)
  Web 表(HTML 表)
  Xero
  Zoho(预览)

备注

连接器标记为“预览”意味着,可以试用它并向我们提供反馈。 若要在解决方案中使用预览版连接器的依赖项,请联系 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 数据工厂支持的其他控制流活动: