Atividade de Pesquisa no Azure Data Factory e no Azure Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

A atividade de pesquisa pode recuperar um conjunto de dados de qualquer uma das fontes de dados com suporte dos pipelines do data factory e do Synapse. Você pode usá-la para determinar dinamicamente em quais objetos operar em uma atividade subsequente, em vez de embutir o nome do objeto. Alguns exemplos de objeto são arquivos e tabelas.

A atividade Lookup lê e retorna o conteúdo de um arquivo de configuração ou tabela. Ela também retorna o resultado da execução de uma consulta ou procedimento armazenado. A saída pode ser um valor singleton ou uma matriz de atributos, que pode ser consumida em uma cópia subsequente, uma transformação ou em atividades de fluxo de controle, como a atividade ForEach.

Criar uma atividade de pesquisa com a interface do usuário

Para usar uma atividade de pesquisa em um pipeline, conclua as seguintes etapas:

1. Procure por Pesquisa no painel Atividades do pipeline e arraste uma atividade de Pesquisa para a tela do pipeline.

2. Selecione a nova atividade de Pesquisa na tela, se ainda não estiver selecionada, e a guia Configurações para editar os detalhes.

   

3. Escolha um conjuntos de dados de origem existente ou selecione o botão Novo para criar um novo.

4. As opções para identificar linhas a incluir do conjuntos de dados de origem variam de acordo com o tipo de conjuntos de dados. O exemplo acima mostra as opções de configuração para um conjuntos de dados de texto delimitado. Abaixo estão exemplos de opções de configuração para um SQL de tabela do Azure e um conjuntos de dados OData.

   

   

Funcionalidades com suporte

Observe o seguinte:

• A atividade de pesquisa pode retornar até 5.000 linhas; se o conjunto de resultados contiver mais registros, as primeiras 5.000 linhas serão retornadas.
• A saída da atividade de Pesquisa aceita até 4 MB de tamanho, a atividade falhará se o tamanho exceder o limite.
• A maior duração da atividade de Pesquisa antes de atingir o tempo limite é de 24 horas.

As seguintes fontes de dados são compatíveis com a atividade Lookup.

Sintaxe

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

Propriedades de tipo

Usar o resultado da atividade de Pesquisa

O resultado de pesquisa é retornado na seção output do resultado da execução de atividade.

• Quando firstRowOnly é definido como true (padrão) , o formato de saída é como mostrado no código a seguir. O resultado da pesquisa fica sob um chave firstRow fixa. Para usar o resultado na atividade subsequente, use o padrão de @{activity('LookupActivity').output.firstRow.table}.

  {
      "firstRow":
      {
          "Id": "1",
          "schema":"dbo",
          "table":"Table1"
      }
  }
  

• Quando firstRowOnly for definido como false (padrão) , o formato de saída será como mostrado no código a seguir. Um campo count indica quantos registros são retornados. Os valores detalhados são exibidos em uma matriz value fixa. Nesse caso, a atividade Lookup é seguida por uma atividade Foreach. Passe a matriz value para o campo items da atividade ForEach usando o padrão de @activity('MyLookupActivity').output.value. Para acessar elementos na matriz value, use a seguinte sintaxe: @{activity('lookupActivity').output.value[zero based index].propertyname}. Um exemplo é @{activity('lookupActivity').output.value[0].schema}.

  {
      "count": "2",
      "value": [
          {
              "Id": "1",
              "schema":"dbo",
              "table":"Table1"
          },
          {
              "Id": "2",
              "schema":"dbo",
              "table":"Table2"
          }
      ]
  } 
  

Exemplo

Neste exemplo, o pipeline contém duas atividades: Lookup e Copy. A atividade Copy copia dados de uma tabela SQL em uma instância do Banco de Dados SQL do Azure para o Armazenamento de Blobs do Azure. O nome da tabela SQL é armazenado em um arquivo JSON em armazenamento de Blob. A atividade Lookup pesquisa o nome da tabela em runtime. O JSON é modificado dinamicamente usando essa abordagem. Não é necessário reimplantar pipelines ou conjuntos de dados.

Este exemplo demonstra a pesquisa apenas para a primeira linha. Para pesquisar por todas as linhas e para encadear os resultados com a atividade ForEach, confira os exemplos em Copiar várias tabelas em massa.

Pipeline

• A atividade Lookup está configurada para usar LookupDataset, que se refere a um local no armazenamento de Blobs do Azure. A atividade Lookup lê o nome da tabela SQL de um arquivo JSON nesse local.
• A atividade Copy usa a saída da atividade Lookup, que é o nome da tabela SQL. A propriedade tableName no SourceDataset está configurada para usar a saída da atividade Lookup. A atividade Copy copia dados da tabela SQL para um local no armazenamento de Blobs do Azure. O local é especificado pela propriedade 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"
    }
}

Conjunto de dados de pesquisa

O conjunto de dados de pesquisa é o arquivo sourcetable.json na pasta de consulta do Armazenamento do Azure especificada pelo tipo AzureBlobStorageLinkedService.

{
    "name": "LookupDataset",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureBlobStorageLinkedService",
            "type": "LinkedServiceReference"
        },
        "annotations": [],
        "type": "Json",
        "typeProperties": {
            "location": {
                "type": "AzureBlobStorageLocation",
                "fileName": "sourcetable.json",
                "container": "lookup"
            }
        }
    }
}

Conjunto de dados de origem da atividade Copy

O conjunto de dados de origem usa a saída da atividade Lookup, que é o nome da tabela SQL. A atividade Copy copia dados dessa tabela SQL para um local no armazenamento de Blobs do Azure. O local é especificado pelo conjunto de dados do coletor.

{
    "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"
            }
        }
    }
}

Conjunto de dados do coletor da atividade Copy

A atividade Copy copia dados da tabela SQL para o arquivo filebylookup.csv na pasta csv no Armazenamento do Azure. O arquivo é especificado pela propriedade 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

Você pode usar os dois tipos de formatos a seguir para o arquivo sourcetable.json.

Conjunto de objetos

{
   "Id":"1",
   "schema":"dbo",
   "table":"Table1"
}
{
   "Id":"2",
   "schema":"dbo",
   "table":"Table2"
}

Matriz de objetos

[ 
    {
        "Id": "1",
        "schema":"dbo",
        "table":"Table1"
    },
    {
        "Id": "2",
        "schema":"dbo",
        "table":"Table2"
    }
]

Limitações e Soluções Alternativas

Aqui estão algumas limitações da atividade de pesquisa e soluções alternativas sugeridas.

Conteúdo relacionado

Confira outras atividades de fluxo de controle com suporte do Azure Data Factory e dos pipelines do Synapse:

• Atividade de execução de pipeline
• Atividade ForEach
• Atividade GetMetadata
• Atividade da Web