Конвейеры и активности в Azure Data Factory и Azure Synapse Analytics

ПРИМЕНИМО К: Azure Data Factory Azure Synapse Analytics

Эта статья поможет понять конвейеры и действия в Azure Data Factory и Azure Synapse Analytics и использовать их для создания комплексных рабочих процессов, управляемых данными, для сценариев перемещения данных и обработки данных.

Обзор

Рабочая область Фабрики данных или Synapse может содержать один или несколько конвейеров. Конвейеры — это логические группы действий, которые вместе отвечают за выполнение задачи. Например, конвейер обработки данных может содержать набор действий, загружающих и очищающих данные журнала, а затем запускающих поток сопоставления данных для их анализа. Потоковая обработка позволяет управлять действиями в виде набора, а не каждым из них в отдельности. Вы развертываете и планируете конвейер, а не действия отдельно.

Действия в конвейере определяют действия, выполняемые с данными. Например, можно использовать действие копирования для копирования данных из SQL Server в Azure Blob Storage. Затем используйте действие потока данных или действие в Databricks Notebook для обработки и преобразования данных из объектного хранилища BLOB в пул Azure Synapse Analytics, на основе которого создаются отчеты бизнес-аналитики.

Azure Data Factory и Azure Synapse Analytics имеют три группировки действий: data movement activities, data transformation activities и control activities. У действия может быть ноль или несколько входных наборов данных, и оно может производить один или несколько выходных наборов данных. На следующей схеме показана связь между конвейером, действием и набором данных:

Входной набор данных представляет входные данные для действия в конвейере, а выходной набор данных — выходные данные для действия. Наборы данных представляют данные в разных хранилищах, например в таблицах, файлах, папках и документах. Созданные наборы данных можно использовать для действий в конвейере. Например, набор данных может быть входным/выходным набором данных для действия копирования или действия HDInsightHive. Дополнительные сведения о наборах данных см. в статье Наборы данных в Azure Data Factory.

Действия перемещения данных

Действие копирования в Фабрике данных копирует данные из хранилища-источника в хранилище-приемник. Фабрика данных поддерживает хранилища данных, перечисленные в таблице в этом разделе. Данные из любого источника можно записывать в любой приемник.

Дополнительные сведения см. в статье Обзор — действие копирования.

Выберите хранилище данных, чтобы узнать, как скопировать данные в него и из него.

Действия преобразования данных

Azure Data Factory и Azure Synapse Analytics поддерживают следующие действия преобразования, которые можно добавить отдельно или связать с другим действием.

Дополнительные сведения см. в статье Преобразование данных.

Действия управления потоком

Поддерживаются следующие действия потока управления:

Создание конвейера с помощью пользовательского интерфейса

JSON-файл конвейера.

Вот как конвейер определен в формате JSON:

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

Активность JSON

В разделе activities можно определить одно или несколько действий. Существует два основных типа действий: действия выполнения и действия управления.

Операции выполнения

Действия выполнения включают в себя действия перемещения данных и преобразования данных. Они имеют следующую структуру верхнего уровня:

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

В приведенной ниже таблице описаны свойства в JSON-определении действий:

Политика действий

Политики влияют на поведение действия во время его выполнения, предоставляя возможность настройки параметров. Политики действий доступны только для выполнения действий.

Определение JSON политики активности (Activity policy JSON definition)

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

Действие управления

Действие управления имеет следующую структуру верхнего уровня:

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

Зависимость активностей

Зависимость действий определяет, как последующие действия зависят от предыдущих, указывая, следует ли продолжать выполнение следующей задачи. Действие может зависеть от одного или нескольких предыдущих действий с разными условиями зависимости.

Различные условия зависимости: "Выполнено", "Сбой", "Пропущено", "Завершено".

Например, если в потоке данных есть активность A -> активность B, возможны различные развития событий:

• Действие B имеет условие зависимости от действия А с условием Выполнено: действие B выполняется, только если действие А имеет конечное состояние "Выполнено".
• Действие B имеет условие зависимости от действия А с условием Сбой: действие B выполняется, только если действие А имеет конечное состояние "Сбой".
• Действие Б имеет условие зависимости от действия А с условием Завершено: действие B выполняется, если действие А имеет конечное состояние "Выполнено" или "Сбой".
• Для действия Б задано условие зависимости от действия А с условием Пропущено: действие B выполняется, если действие А находится в окончательном состоянии "Пропущено". Состояние "Пропущено" возникает в сценарии, когда действие X -> действие Y -> действие Z, и каждое действие выполняется только в случае успешного завершения предыдущего действия. При сбое действия X действие Y остается в состоянии "Пропущено", так как оно никогда не выполнится. Аналогично, действие Z находится в состоянии "Пропущено".

Пример. Действие 2 зависит от успешного завершения действия 1

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

Пример конвейера копирования

В следующем примере конвейера в разделе activities есть одно действие типа Copy. В этом примере действие copy копирует данные из хранилища BLOB-объектов Azure в базу данных в Azure SQL Database.

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

Обратите внимание на следующие аспекты:

• В разделе действий есть только одно действие, тип которого имеет значение Copy.
• Для этого действия параметру input присвоено значение InputDataset, а параметру output — значение OutputDataset. Сведения об определении наборов данных в JSON см. в статье Наборы данных.
• В разделе typeProperties в качестве типа источника указано BlobSource, а в качестве типа приемника — SqlSink. В разделе действий по перемещению данных выберите хранилище данных, которое вы хотите использовать в качестве источника или приемника, чтобы узнать больше о перемещении данных в это хранилище данных или из нее.

Полное пошаговое руководство по созданию этого конвейера см. в статье Краткое руководство: создание фабрики данных.

Пример конвейера преобразования

В следующем примере конвейера в разделе activities есть одно действие типа HDInsightHive. В этом примере действие HDInsight Hive преобразует данные из хранилища BLOB-объектов Azure путём выполнения скрипта Hive в кластере Azure HDInsight Hadoop.

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

Обратите внимание на следующие аспекты:

• В разделе действий есть только одно действие, тип которого имеет значение HDInsightHive.
• Файл скрипта Hive, partitionweblogs.hql, хранится в учетной записи Azure Storage (указанной скриптомLinkedService с именем AzureStorageLinkedService) и в папке скрипта в контейнере adfgetstarted.
• Раздел defines используется, чтобы указать параметры среды выполнения, передаваемые скрипту Hive в качестве значений конфигурации Hive (например, {hiveconf:inputtable}, $${hiveconf:partitionedtable}).

Разделы typeProperties для каждого действия преобразования отличаются. Чтобы узнать о свойствах типов, поддерживаемых для действия преобразования, выберите действие преобразования в действиях преобразования данных.

Полное пошаговое руководство по созданию этого конвейера см. в статье Учебник: преобразование данных с использованием Spark.

Несколько действий в конвейере

В двух предыдущих примерах конвейеров содержится только одно действие. Конвейер может включать несколько задач. Если в конвейере есть несколько действий, а последующие действия не зависят от предыдущих действий, эти действия могут выполняться параллельно.

Два действия можно соединить с помощью зависимости действий, определяющей зависимость последующих действий от предыдущих, а также то, следует ли продолжать выполнение следующей задачи. Действие может зависеть от одного или нескольких предыдущих действий с разными условиями зависимости.

Планирование конвейеров

Конвейеры планируются триггерами. Существуют различные типы триггеров (триггер планировщика, который позволяет запускать конвейеры по расписанию, а также ручной триггер, который активирует конвейеры по запросу). Дополнительные сведения см. в статье Pipeline execution and triggers in Azure Data Factory (Выполнение конвейера и триггеры в фабрике данных Azure).

Чтобы триггер запускал выполнение конвейера, необходимо включить в определение триггера ссылку на конкретный конвейер. Конвейеры и триггеры имеют связь n-m. Один триггер может запускать несколько конвейеров, и несколько триггеров могут запускать один конвейер. После определения триггера необходимо активировать его, чтобы он начал запуск конвейера. Дополнительные сведения см. в статье Pipeline execution and triggers in Azure Data Factory (Выполнение конвейера и триггеры в фабрике данных Azure).

Предположим, у вас есть триггер по расписанию («Триггер A»), который я хочу использовать для запуска конвейера «MyCopyPipeline». Вы определяете триггер, как показано в следующем примере:

Определение триггера А

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

Связанный контент

• Постройка конвейера с активностью копирования
• Создание конвейера с активностью преобразования данных
• Как реализовать CI/CD (непрерывную интеграцию и развертывание) с помощью Azure Data Factory