Pipelines e atividades em Azure Data Factory e Azure Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Este artigo ajuda-o a compreender pipelines e atividades no Azure Data Factory e no Azure Synapse Analytics e a usá-los para construir fluxos de trabalho orientados por dados de ponta a ponta para os seus cenários de movimentação e processamento de dados.

Descrição geral

Um Data Factory ou Synapse Workspace pode ter um ou mais pipelines. Um pipeline é um agrupamento lógico de atividades que, em conjunto, executam uma tarefa. Por exemplo, um pipeline pode conter um conjunto de atividades que ingerem e limpam dados de registos e, em seguida, iniciam um fluxo de dados de mapeamento para analisar os dados de registo. O pipeline permite-lhe gerir as atividades como um conjunto, em vez de cada uma individualmente. Em vez de implementar e agendar as atividades de forma independente, implementa e agenda o pipeline.

As atividades num pipeline definem as ações a executar nos seus dados. Por exemplo, pode usar uma atividade de cópia para copiar dados do SQL Server para um Armazenamento de Blobs do Azure. Depois, utilize uma atividade de fluxo de dados ou uma atividade Databricks Notebook para processar e transformar dados do armazenamento de blob para um pool Azure Synapse Analytics, sobre o qual são construídas soluções de relatórios de business intelligence.

Azure Data Factory e Azure Synapse Analytics têm três agrupamentos de atividades: atividades de movimento de dados, atividades de transformação de dados, e atividades de controlo. Uma atividade pode ter zero ou mais conjuntos de dados de entrada e produzir um ou mais conjuntos de dados de saída. O diagrama a seguir mostra a relação entre pipeline, atividade e conjunto de dados:

Um conjunto de dados de entrada representa a entrada para uma atividade no pipeline e um conjunto de dados de saída representa a saída para a atividade. Os conjuntos de dados identificam dados dentro de diferentes arquivos de dados, como tabelas, ficheiros, pastas e documentos. Depois de criar um conjunto de dados, pode utilizá-lo com atividades num pipeline. Por exemplo, um conjunto de dados pode ser um conjunto de dados de entrada/saída de uma atividade Cópia ou de uma atividade HDInsightHive. Para mais informações sobre conjuntos de dados, consulte o artigo Datasets in Azure Data Factory.

Atividades de movimento de dados

Atividade de Cópia do Data Factory copia dados de um repositório de dados de origem para um repositório de dados de destino. O Data Factory suporta os arquivos de dados apresentados na tabela nesta secção. Os dados de qualquer origem podem ser escritos em qualquer sink.

Para obter mais informações, veja o artigo Copy Activity - Overview (Atividade de Cópia - Descrição Geral).

Selecione um arquivo de dados para saber como copiar dados de e para esse arquivo.

Atividades de transformação de dados

Azure Data Factory e Azure Synapse Analytics suportam as seguintes atividades de transformação que podem ser adicionadas individualmente ou em cadeia com outra atividade.

Para obter mais informações, veja o artigo Data Transformation Activities (Atividades de Transformação de Dados).

Atividades de controle de fluxo

São suportadas as atividades de fluxo de controlo abaixo:

Criando um pipeline com a interface do usuário

JSON do Pipeline

Veja como um pipeline é definido no formato JSON:

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities":
        [
        ],
        "parameters": {
        },
        "concurrency": <your max pipeline concurrency>,
        "annotations": [
        ]
    }
}

JSON da Atividade

A secção atividades pode ter uma ou mais atividades definidas na mesma. Existem dois tipos principais de atividades: atividades de Execução e de Controlo.

Atividades de Execução

As atividades de execução incluem atividades de movimento de dados e de transformação de dados. Têm a estrutura de nível superior abaixo:

{
    "name": "Execution Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "linkedServiceName": "MyLinkedService",
    "policy":
    {
    },
    "dependsOn":
    {
    }
}

A tabela seguinte descreve as propriedades na definição JSON da atividade:

Política de atividade

As políticas afetam o comportamento em tempo de execução de uma atividade, fornecendo opções de configuração. Políticas de Atividade apenas estão disponíveis para atividades de execução.

Definição JSON da política de atividade

{
    "name": "MyPipelineName",
    "properties": {
      "activities": [
        {
          "name": "MyCopyBlobtoSqlActivity",
          "type": "Copy",
          "typeProperties": {
            ...
          },
         "policy": {
            "timeout": "00:10:00",
            "retry": 1,
            "retryIntervalInSeconds": 60,
            "secureOutput": true
         }
        }
      ],
        "parameters": {
           ...
        }
    }
}

Atividade de controlo

As atividades de controlo têm a estrutura de nível superior seguinte:

{
    "name": "Control Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "dependsOn":
    {
    }
}

Dependência das atividades

A Dependência de Atividade define como as atividades subsequentes dependem de atividades anteriores, determinando a condição de continuar ou não executando a próxima tarefa. Uma atividade pode depender de uma ou várias atividades anteriores com condições de dependência diferentes.

As condições de dependência diferentes são: Bem-sucedidas, Falhas, Ignoradas, Concluídas.

Por exemplo, se um pipeline tiver a Atividade A -> Atividade B, os diferentes cenários que podem ocorrer são:

• A Atividade B tem a condição de dependência de a Atividade A ser bem-sucedida - a Atividade B só é executada se o estado final de A for “bem-sucedida”
• A Atividade B tem a condição de dependência de a Atividade A falhar - a Atividade B só é executada se o estado final de A for “falha”
• A Atividade B depende da condição da Atividade A estar concluída: a Atividade B é executada se o estado final de A for “sucedido” ou “falhado”.
• A Atividade B tem uma condição de dependência da Atividade A com ignorado: A Atividade B é executada se a Atividade A tiver um status final de ignorado. Ignorado ocorre no cenário de Atividade X -> Atividade Y -> Atividade Z, onde cada atividade é executada somente se a atividade anterior for bem-sucedida. Se a Atividade X falhar, a Atividade Y terá um status de "Ignorado" porque nunca será executada. Da mesma forma, a Atividade Z também tem um status de "Ignorado".

Exemplo: a Atividade 2 depende de a Atividade 1 ser bem-sucedida

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities": [
         {
            "name": "MyFirstActivity",
            "type": "Copy",
            "typeProperties": {
            },
            "linkedServiceName": {
            }
        },
        {
            "name": "MySecondActivity",
            "type": "Copy",
            "typeProperties": {
            },
            "linkedServiceName": {
            },
            "dependsOn": [
            {
                "activity": "MyFirstActivity",
                "dependencyConditions": [
                    "Succeeded"
                ]
            }
          ]
        }
      ],
      "parameters": {
       }
    }
}

Fluxo de cópia de amostra

No pipeline de exemplo a seguir, há uma atividade do tipo Copy na secção de atividades. Neste exemplo, a atividade copy copia dados de um armazenamento de Azure Blob para uma base de dados em Base de Dados SQL do Azure.

{
  "name": "CopyPipeline",
  "properties": {
    "description": "Copy data from a blob to Azure SQL table",
    "activities": [
      {
        "name": "CopyFromBlobToSQL",
        "type": "Copy",
        "inputs": [
          {
            "name": "InputDataset"
          }
        ],
        "outputs": [
          {
            "name": "OutputDataset"
          }
        ],
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "SqlSink",
            "writeBatchSize": 10000,
            "writeBatchTimeout": "60:00:00"
          }
        },
        "policy": {
          "retry": 2,
          "timeout": "01:00:00"
        }
      }
    ]
  }
}

Tenha em conta os seguintes pontos:

• Na seção de atividades, há apenas uma atividade cujo tipo está definido como Copiar.
• A entrada da atividade está definida como InputDataset e a saída como OutputDataset. Veja o artigo Conjuntos de dados para saber como definir conjuntos de dados em JSON.
• Na secção typeProperties, BlobSource está especificado como o tipo de origem e SqlSink como o tipo de sink. Na seção de atividades de movimentação de dados, selecione o armazenamento de dados que você deseja usar como fonte ou coletor para saber mais sobre como mover dados de/para esse armazenamento de dados.

Para obter um passo a passo completo sobre como criar esse pipeline, consulte Guia de início rápido: criar um Data Factory.

Pipeline de transformação de amostra

No pipeline de exemplo a seguir, há uma atividade do tipo HDInsightHive na seção de atividades . Neste exemplo, a atividade Hive HDInsight transforma dados de um armazenamento Azure Blob ao executar um ficheiro de script Hive num cluster Hadoop Azure HDInsight.

{
    "name": "TransformPipeline",
    "properties": {
        "description": "My first Azure Data Factory pipeline",
        "activities": [
            {
                "type": "HDInsightHive",
                "typeProperties": {
                    "scriptPath": "adfgetstarted/script/partitionweblogs.hql",
                    "scriptLinkedService": "AzureStorageLinkedService",
                    "defines": {
                        "inputtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.windows.net/inputdata",
                        "partitionedtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.windows.net/partitioneddata"
                    }
                },
                "inputs": [
                    {
                        "name": "AzureBlobInput"
                    }
                ],
                "outputs": [
                    {
                        "name": "AzureBlobOutput"
                    }
                ],
                "policy": {
                    "retry": 3
                },
                "name": "RunSampleHiveActivity",
                "linkedServiceName": "HDInsightOnDemandLinkedService"
            }
        ]
    }
}

Tenha em conta os seguintes pontos:

• Na seção de atividades, há apenas uma atividade cujo tipo está definido como HDInsightHive.
• O ficheiro de script Hive, partitionweblogs.hql, é armazenado na conta Armazenamento do Azure (especificada pelo scriptLinkedService, chamado AzureStorageLinkedService) e na pasta de scripts no contentor adfgetstarted.
• A secção defines é utilizada para especificar as definições de runtime que são transmitidas ao script do Hive como valores de configuração do Hive (por exemplo, ${hiveconf:inputtable} e ${hiveconf:partitionedtable}).

A secção typeProperties é diferente para cada atividade de transformação. Para saber mais sobre as propriedades de tipo suportadas para uma atividade de transformação, selecione a atividade de transformação nas atividades de transformação de dados.

Para obter instruções completas para criar este pipeline, veja Tutorial: transformar dados com o Spark.

Múltiplas atividades num pipeline

Os dois pipelines de exemplo anteriores contêm apenas uma atividade. Pode haver mais do que uma atividade num pipeline. Se você tiver várias atividades em um pipeline e as atividades subsequentes não dependerem de atividades anteriores, as atividades poderão ser executadas em paralelo.

Pode utilizar a dependência das atividades para encadear duas atividades; a dependência define de que forma é que as atividades subsequentes dependem das atividades anteriores, determinando a condição de continuação da execução da tarefa seguinte. Uma atividade pode depender de uma ou várias atividades anteriores com condições de dependência diferentes.

Agendamento de pipelines

Os pipelines são agendados por acionadores. Existem diferentes tipos de gatilhos (gatilho de agenda, que permite que pipelines sejam acionados num cronograma de tempo fixo, e o gatilho manual, que aciona pipelines sob demanda). Para obter mais informações sobre os acionadores, veja o artigo Pipeline execution and triggers (Execução e acionadores de pipelines).

Para que o seu acionador inicie uma execução de pipeline, é necessário incluir uma referência ao pipeline específico na definição do acionador. Os pipelines e os acionadores têm uma relação “n-m” (muitos para muitos). Múltiplos acionadores podem iniciar um único pipeline, e o mesmo acionador pode iniciar vários pipelines. Quando o acionador estiver definido, tem de iniciá-lo para que o mesmo comece a acionar o pipeline. Para obter mais informações sobre os acionadores, veja o artigo Pipeline execution and triggers (Execução e acionadores de pipelines).

Por exemplo, imagina que tens um gatilho do Scheduler, "Gatilho A", para iniciar o teu pipeline, "MyCopyPipeline". Define o gatilho, conforme mostrado no exemplo a seguir:

Definição do Acionador A

{
  "name": "TriggerA",
  "properties": {
    "type": "ScheduleTrigger",
    "typeProperties": {
      ...
      }
    },
    "pipeline": {
      "pipelineReference": {
        "type": "PipelineReference",
        "referenceName": "MyCopyPipeline"
      },
      "parameters": {
        "copySourceName": "FileSource"
      }
    }
  }
}

Conteúdos relacionados

• Criar um pipeline com uma tarefa de cópia
• Criar um pipeline com uma atividade de transformação de dados
• Como alcançar CI/CD (integração e entrega contínua) usando Azure Data Factory