Atividade de Fluxo de Dados no Azure Data Factory e no Azure Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Use a atividade Fluxo de Dados para transformar e mover dados por meio de fluxos de dados de mapeamento. Se você for novo com fluxos de dados, confira Visão geral do fluxo de dados de mapeamento

Criar uma atividade de Fluxo de Dados com a interface do usuário

Para usar uma atividade Fluxo de Dados em um pipeline, conclua as etapas a seguir:

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

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

   

3. A chave do ponto de verificação é usada para definir o ponto de verificação quando o fluxo de dados é usado para captura de dados de alterações. Você pode substituí-la. As atividades de fluxo de dados usam um valor de guid como chave do ponto de verificação em vez de "nome do pipeline + nome da atividade", para que sempre possam continuar acompanhando o estado da captura de dados de alterações do cliente, mesmo que haja ações de renomeação. Todas as atividades de fluxo de dados existentes usam a chave de padrão antiga para fins de compatibilidade com versões anteriores. A opção de chave de ponto de verificação depois de publicar uma nova atividade de fluxo de dados com o recurso de fluxo de dados habilitado para captura de dados de alterações é mostrada abaixo.

   

4. Selecione um fluxo de dados existente ou crie um novo usando o botão Novo. Selecione outras opções conforme necessário para concluir sua configuração.

Sintaxe

{
    "name": "MyDataFlowActivity",
    "type": "ExecuteDataFlow",
    "typeProperties": {
      "dataflow": {
         "referenceName": "MyDataFlow",
         "type": "DataFlowReference"
      },
      "compute": {
         "coreCount": 8,
         "computeType": "General"
      },
      "traceLevel": "Fine",
      "runConcurrently": true,
      "continueOnError": true,      
      "staging": {
          "linkedService": {
              "referenceName": "MyStagingLinkedService",
              "type": "LinkedServiceReference"
          },
          "folderPath": "my-container/my-folder"
      },
      "integrationRuntime": {
          "referenceName": "MyDataFlowIntegrationRuntime",
          "type": "IntegrationRuntimeReference"
      }
}

Propriedades de tipo

Dimensionar dinamicamente a computação de fluxo de dados em runtime

As propriedades Core Count e Compute Type podem ser definidas dinamicamente para ajustar o tamanho de seus dados de origem de entrada em runtime. Use atividades de pipeline como Lookup ou Get Metadata para localizar o tamanho dos dados do conjunto de dados de origem. Em seguida, use Add Dynamic Content nas propriedades de atividade do fluxo de dados. Você pode escolher tamanhos de computação pequenos, médios ou grandes. Opcionalmente, escolha "Personalizado" e configure os tipos de computação e o número de núcleos manualmente.

Este é um breve tutorial em vídeo explicando essa técnica

Runtime de integração do fluxo de dados

Escolha qual runtime de integração usar para a execução da atividade do fluxo de dados. Por padrão, o serviço usa o runtime de integração de resolução automática do Azure com quatro núcleos de trabalho. Esse IR tem um tipo de computação de uso geral e é executado na mesma região que a sua instância de serviço. Para pipelines operacionalizados, é altamente recomendável que você crie seus próprios Azure Integration Runtimes que definem regiões específicas, tipo de computação, contagens principais e TTL para a execução da atividade de fluxo de dados.

Um tipo de computação mínimo de Uso Geral com uma configuração de 8 + 8 (16 núcleos virtuais no total), e uma TTL (vida útil) de dez minutos é a recomendação mínima para a maioria das cargas de trabalho de produção. Ao definir um pequeno TTL, o Azure IR pode manter um cluster quente que não incorrerá em vários minutos de tempo de início para um cluster frio. Para obter mais informações, confira Azure Integration Runtime.

PolyBase

Se você estiver usando um Azure Synapse Analytics como um coletor ou origem, deverá escolher um local de preparo para o carregamento em lotes do PolyBase. O PolyBase permite o carregamento em lotes em massa em vez de carregar os dados linha por linha. O PolyBase reduz bastante o tempo de carregamento no Azure Synapse Analytics.

Chave do ponto de verificação

Ao usar a opção de captura de alterações para fontes de fluxo de dados, o ADF mantém e gerencia o ponto de verificação para você automaticamente. A chave de ponto de verificação padrão é um hash do nome do fluxo de dados e do nome do pipeline. Se você estiver usando um padrão dinâmico para suas tabelas ou pastas de origem, talvez queira substituir esse hash e definir seu valor de chave de ponto de verificação aqui.

Nível de log

Se não precisar de cada execução de pipeline das suas atividades de fluxo de dados para registrar completamente todos os logs de telemetria detalhados, você pode definir seu nível de registros em log como "Básico" ou "Nenhum". Ao executar os fluxos de dados no modo “Detalhado” (padrão), você solicita que o serviço registre totalmente a atividade em cada nível de partição individual durante a transformação dos dados. Isso pode ser uma operação dispendiosa. Portanto, habilitar o modo detalhado somente ao solucionar problemas pode melhorar o fluxo geral de dados e o desempenho do pipeline. O modo "Básico" só registra as durações de transformação e "Nenhum" fornece apenas um resumo das durações.

Propriedades do coletor

O recurso de agrupamento em fluxos de dados permite que você defina a ordem de execução de seus coletores, bem como dos coletores de grupo juntos usando o mesmo número de grupo. Para ajudar a gerenciar os grupos, você pode pedir que o serviço execute coletores no mesmo grupo em paralelo. Você também pode definir o grupo de coletores para continuar mesmo depois que um deles encontrar um erro.

O comportamento padrão dos coletores de fluxo de dados é executar cada coletor sequencialmente, em série, e fazer com que o fluxo de dados falhe quando um erro for encontrado no coletor. Além disso, todos os coletores são padronizados para o mesmo grupo, a menos que você acesse as propriedades de fluxo de dados e defina prioridades diferentes para os coletores.

Primeira linha apenas

Essa opção só está disponível para fluxos de dados que têm coletores de cache habilitados para "Saída para atividade". A saída do fluxo de dados que é injetada diretamente em seu pipeline é limitada a 2 MB. A configuração "primeira linha somente" ajuda a limitar a saída de dados do fluxo de dados ao injetar a saída da atividade de fluxo de dados diretamente em seu pipeline.

Parametrização de fluxos de dados

Conjuntos de dados parametrizados

Se o fluxo de dados usar conjuntos de dados parametrizados, defina os valores de parâmetro na guia Configurações.

Fluxos de dados parametrizados

Se o fluxo de dados for parametrizado, defina os valores dinâmicos dos parâmetros de fluxo de dados na guia Parâmetros. Você pode usar tanto a linguagem de expressão de pipeline como a linguagem de expressão do fluxo de dados para atribuir valores de parâmetros dinâmicos ou literais. Para obter mais informações, confira Parâmetros de Fluxo de Dados.

Propriedades de computação parametrizadas.

Você pode parametrizar a contagem de núcleos ou o tipo de computação se usar o Azure Integration Runtime de resolução automática e especificar valores para compute.coreCount e compute.computeType.

Depuração de pipeline da atividade de Fluxo de Dados

Para uma execução de pipeline de depuração com uma atividade de Fluxo de Dados, você deve alternar o modo de depuração de fluxo de dados por meio do controle deslizante Depuração de Fluxo de Dados na barra superior. O modo de depuração permite executar o fluxo de dados em um cluster do Spark ativo. Para saber mais, consulte Modo de depuração.

O pipeline de depuração é executado no cluster de depuração ativa, não no ambiente do runtime de integração especificado nas configurações de atividade do Fluxo de Dados. Você pode escolher o ambiente de computação de depuração ao iniciar o modo de depuração.

Monitoramento da atividade de Fluxo de Dados

A atividade de Fluxo de Dados tem uma experiência de monitoramento especial, na qual você pode exibir informações sobre particionamento, tempo de preparo e linhagem de dados. Abra o painel de monitoramento usando o ícone de óculos em Ações. Para saber mais, confira Monitoramento de fluxos de dados.

Usar resultados da atividade de Fluxo de Dados em uma atividade subsequente

A atividade de fluxo de dados gera métricas referentes ao número de linhas gravadas em cada coletor e linhas lidas de cada fonte. Esses resultados são retornados na seção output do resultado da execução de atividade. As métricas retornadas estão no formato do JSON abaixo.

{
    "runStatus": {
        "metrics": {
            "<your sink name1>": {
                "rowsWritten": <number of rows written>,
                "sinkProcessingTime": <sink processing time in ms>,
                "sources": {
                    "<your source name1>": {
                        "rowsRead": <number of rows read>
                    },
                    "<your source name2>": {
                        "rowsRead": <number of rows read>
                    },
                    ...
                }
            },
            "<your sink name2>": {
                ...
            },
            ...
        }
    }
}

Por exemplo, para obter o número de linhas gravadas em um coletor chamado 'sink1' em uma atividade denominada 'dataflowActivity', use @activity('dataflowActivity').output.runStatus.metrics.sink1.rowsWritten.

Para obter o número de linhas lidas de uma fonte denominada 'source1' que foi utilizada nesse coletor, use @activity('dataflowActivity').output.runStatus.metrics.sink1.sources.source1.rowsRead.

Conteúdo relacionado

Confira as atividades de fluxo de controle compatíveis:

• Atividade Condição Se
• Atividade de execução de pipeline
• Para cada atividade
• Atividade de obtenção de metadados
• Atividade de pesquisa
• Atividade da Web
• Atividade Until