Использование настраиваемых активностей в рамках конвейера Azure Data Factory или Azure Synapse Analytics

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

Совет

Data Factory в Microsoft Fabric — это следующее поколение Azure Data Factory с более простой архитектурой, встроенным ИИ и новыми функциями. Если вы не знакомы с интеграцией данных, начните с Fabric Data Factory. Существующие рабочие нагрузки ADF могут обновляться до Fabric для доступа к новым возможностям в области обработки и анализа данных, аналитики в режиме реального времени и отчетов.

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

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

Чтобы переместить данные в хранилище данных, которое не поддерживает служба, и обратно, или чтобы преобразовать или обработать данные способом, который не поддерживается службой, создайте пользовательское действие с собственной логикой перемещения или преобразования данных и используйте это действие в конвейере. Настраиваемое действие выполняет настраиваемую логику кода в пуле виртуальных машин Azure Batch.

Примечание.

Мы рекомендуем использовать модуль Az PowerShell Azure для взаимодействия с Azure. Сведения о начале работы см. в разделе Install Azure PowerShell. Сведения о миграции в модуль Az PowerShell см. в статье Migrate Azure PowerShell из AzureRM в Az.

Ознакомьтесь со следующими статьями, если вы не знакомы с службой Azure Batch:

Azure Batch Basics чтобы получить обзор службы Azure Batch.
командлет New-AzBatchAccount для создания учетной записи Azure Batch или портал Azure для создания учетной записи Azure Batch через портал Azure. Подробные инструкции по использованию командлета см. в статье Использование PowerShell для управления учетной записью Azure Batch.
New-AzBatchPool cmdlet для создания пула Azure Batch.

Внимание

При создании нового пула Azure Batch необходимо использовать "VirtualMachineConfiguration" и НЕ "CloudServiceConfiguration".

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

Чтобы использовать настраиваемую активность в конвейере, выполните следующие шаги:

Найдите элемент Настраиваемое на панели конвейера "Действия" и перетащите его на холст конвейера.
Выберите новое настраиваемое действие на холсте, если оно еще не выбрано.
Перейдите на вкладку Azure Batch, чтобы выбрать или создать связанную службу Azure Batch, которая будет выполнять настраиваемое действие.
Перейдите на вкладку Settings и укажите команду, выполняемую в Azure Batch, и дополнительные сведения.

связанная служба Azure Batch

Следующий код JSON определяет пример связанной службы Azure Batch. Подробнее см. Поддерживаемые вычислительные среды

{
    "name": "AzureBatchLinkedService",
    "properties": {
        "type": "AzureBatch",
        "typeProperties": {
            "accountName": "batchaccount",
            "accessKey": {
                "type": "SecureString",
                "value": "access key"
            },
            "batchUri": "https://batchaccount.region.batch.azure.com",
            "poolName": "poolname",
            "linkedServiceName": {
                "referenceName": "StorageLinkedService",
                "type": "LinkedServiceReference"
            }
        }
    }
}

Дополнительные сведения о связанной службе Azure Batch см. в статье Compute linked services.

Настраиваемое действие

В следующем фрагменте кода JSON определяется конвейер с простым настраиваемым действием. Определение действия содержит ссылку на связанную службу Azure Batch.

{
  "name": "MyCustomActivityPipeline",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "helloworld.exe",
        "folderPath": "customactv2/helloworld",
        "resourceLinkedService": {
          "referenceName": "StorageLinkedService",
          "type": "LinkedServiceReference"
        }
      }
    }]
  }
}

В этом примере helloworld.exe — это пользовательское приложение, хранящееся в папке customactv2/helloworld учетной записи Azure Storage, используемой в resourceLinkedService. Пользовательская активность отправляет это пользовательское приложение для выполнения на Azure Batch. Вы можете заменить команду на любое предпочтительное приложение, которое можно выполнить в целевой операционной системе узлов Azure Batch пула.

В следующей таблице описаны имена и описания свойств, относящихся к этому действию.

Свойство	Описание:	Обязательное поле
имя	Название активности в конвейере задач.	Да
описание	Описание того, что делает действие.	Нет
тип	Для пользовательского действия используется тип действия Custom.	Да
ИмяСвязанногоСервиса	Связанная служба с Azure Batch. Чтобы узнать больше об этой связанной службе, см. статью Связанные вычислительные службы.	Да
Команда	Команда пользовательского приложения для выполнения. Если приложение уже доступно на узле пула Azure Batch, можно пропустить ресурсLinkedService и folderPath. Например, можно указать команду, которая должна быть `cmd /c dir`, которая изначально поддерживается узлом Windows пакетного пула.	Да
resourceLinkedService	Служба Azure Storage, связанная с учетной записью Azure Storage, в которой размещено пользовательское приложение	Нет*
folderPath	Путь к папке пользовательского приложения и всех его зависимостей. Если у вас есть зависимости, хранящиеся в вложенных папках, то есть в иерархической структуре папок в разделе folderPath — структура папок в настоящее время сглаживается при копировании файлов в Azure Batch. То есть все файлы копируются в одну папку без подпапок. Чтобы обойти это поведение, рекомендуется рассмотреть возможность сжатия файлов, копирования сжатого файла, а затем распаковки его, используя пользовательский код, в нужном месте.	Нет*
опорные объекты	Массив имеющихся связанных служб и наборов данных. Указанные связанные службы и наборы данных передаются в пользовательское приложение в формате JSON. Пользовательский код может использовать ресурсы службы	Нет
расширенные свойства	Определенные пользователем свойства, которые могут быть переданы в пользовательское приложение в формате JSON. Пользовательский код может использовать дополнительные свойства.	Нет
Время хранения в днях	Время хранения файлов, отправленных для настраиваемого действия. Значение по умолчанию — 30 суток.	Нет

* Свойства resourceLinkedService и folderPath (оба) должны быть указаны либо пропущены.

Примечание.

Если вы передаете связанные службы как объекты ссылок в пользовательском действии, рекомендуется передать связанную службу с поддержкой функций Azure Key Vault (так как она не содержит конфиденциальных строк) и получить учетные данные с помощью имени секрета непосредственно из кода, используя Key Vault. Вы можете найти пример здесь, который ссылается на подключенную службу с поддержкой AKV, извлекает учетные данные из Key Vault, а затем получает доступ к хранилищу в коде.

Примечание.

В настоящее время поддерживается только Azure Blob Storage для связанной службы ресурсов в настраиваемой активности, и это единственная связанная служба, которая создается по умолчанию. Нет возможности выбрать другие коннекторы, такие как ADLS Gen2.

Разрешения настраиваемых действий

Настраиваемое действие задает для учетной записи авто-пользователя Azure Batch права доступа не администратора с областью задач (спецификация авто-пользователя по умолчанию). Уровень разрешений автоматической учетной записи пользователя изменить невозможно. Дополнительные сведения см. в разделе статьи "Выполнение задач с учетными записями пользователей в пакетных операциях | Автоматические учетные записи пользователей".

Выполнение команд

Вы можете непосредственно выполнить команду, используя настраиваемое действие. В следующем примере выполняется команда "echo hello world" на конечных узлах пула Azure Batch, и вывод отображается в stdout.

{
  "name": "MyCustomActivity",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "cmd /c echo hello world"
      }
    }]
  }
}

Передача объектов и свойств

В этом примере показано, как использовать свойства referenceObjects и extendedProperties для передачи объектов и определенных пользователем свойств из службы в пользовательское приложение.

{
  "name": "MyCustomActivityPipeline",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "SampleApp.exe",
        "folderPath": "customactv2/SampleApp",
        "resourceLinkedService": {
          "referenceName": "StorageLinkedService",
          "type": "LinkedServiceReference"
        },
        "referenceObjects": {
          "linkedServices": [{
            "referenceName": "AzureBatchLinkedService",
            "type": "LinkedServiceReference"
          }]
        },
        "extendedProperties": {          
          "connectionString": {
            "type": "SecureString",
            "value": "aSampleSecureString"
          },
          "PropertyBagPropertyName1": "PropertyBagValue1",
          "propertyBagPropertyName2": "PropertyBagValue2",
          "dateTime1": "2015-04-12T12:13:14Z"
        }
      }
    }]
  }
}

При выполнении действия referenceObjects и extendedProperties сохраняются в указанных ниже файлах. Эти файлы будут развернуты в той же папке выполнения SampleApp.exe.

activity.json

Хранит свойство extendedProperties и свойства настраиваемых действий.
linkedServices.json

Хранит массив связанных служб, определенных в свойстве referenceObjects.
datasets.json

Хранит массив наборов данных, определенных в свойстве referenceObjects.

В следующем примере кода показано, как SampleApp.exe может получить доступ к необходимой информации в JSON-файлах.

using Newtonsoft.Json;
using System;
using System.IO;

namespace SampleApp
{
    class Program
    {
        static void Main(string[] args)
        {
            //From Extend Properties
            dynamic activity = JsonConvert.DeserializeObject(File.ReadAllText("activity.json"));
            Console.WriteLine(activity.typeProperties.extendedProperties.connectionString.value);

            // From LinkedServices
            dynamic linkedServices = JsonConvert.DeserializeObject(File.ReadAllText("linkedServices.json"));
            Console.WriteLine(linkedServices[0].properties.typeProperties.accountName);
        }
    }
}

Извлечение выходных данных выполнения

Чтобы запустить выполнение конвейера, выполните следующую команду PowerShell:

$runId = Invoke-AzDataFactoryV2Pipeline -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineName $pipelineName

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

while ($True) {
    $result = Get-AzDataFactoryV2ActivityRun -DataFactoryName $dataFactoryName -ResourceGroupName $resourceGroupName -PipelineRunId $runId -RunStartedAfter (Get-Date).AddMinutes(-30) -RunStartedBefore (Get-Date).AddMinutes(30)

    if(!$result) {
        Write-Host "Waiting for pipeline to start..." -foregroundcolor "Yellow"
    }
    elseif (($result | Where-Object { $_.Status -eq "InProgress" } | Measure-Object).count -ne 0) {
        Write-Host "Pipeline run status: In Progress" -foregroundcolor "Yellow"
    }
    else {
        Write-Host "Pipeline '"$pipelineName"' run finished. Result:" -foregroundcolor "Yellow"
        $result
        break
    }
    ($result | Format-List | Out-String)
    Start-Sleep -Seconds 15
}

Write-Host "Activity `Output` section:" -foregroundcolor "Yellow"
$result.Output -join "`r`n"

Write-Host "Activity `Error` section:" -foregroundcolor "Yellow"
$result.Error -join "`r`n"

Выводы stdout и stderr вашего пользовательского приложения сохраняются в контейнере adfjobs в связанной службе Azure Storage, которую вы определили при создании связанной службы Azure Batch с идентификатором GUID задачи. Вы можете получить подробный путь из выходных данных выполнения действия, как показано в следующем фрагменте кода:

Pipeline ' MyCustomActivity' run finished. Result:

ResourceGroupName : resourcegroupname
DataFactoryName   : datafactoryname
ActivityName      : MyCustomActivity
PipelineRunId     : xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
PipelineName      : MyCustomActivity
Input             : {command}
Output            : {exitcode, outputs, effectiveIntegrationRuntime}
LinkedServiceName :
ActivityRunStart  : 10/5/2017 3:33:06 PM
ActivityRunEnd    : 10/5/2017 3:33:28 PM
DurationInMs      : 21203
Status            : Succeeded
Error             : {errorCode, message, failureType, target}

Activity Output section:
"exitcode": 0
"outputs": [
  "https://<container>.blob.core.windows.net/adfjobs/<GUID>/output/stdout.txt",
  "https://<container>.blob.core.windows.net/adfjobs/<GUID>/output/stderr.txt"
]
"effectiveIntegrationRuntime": "DefaultIntegrationRuntime (East US)"
Activity Error section:
"errorCode": ""
"message": ""
"failureType": ""
"target": "MyCustomActivity"

Если вы хотите использовать содержимое stdout.txt в последующих действиях, путь к файлу stdout.txt можно получить в значении выражения "@activity('MyCustomActivity').output.outputs[0]".

Внимание

Свойства activity.json, linkedServices.json и datasets.json хранятся в папке среды выполнения пакетной задачи. Для этого примера файлы activity.json, linkedServices.json и datasets.json хранятся по адресу https://adfv2storage.blob.core.windows.net/adfjobs/<GUID>/runtime/. При необходимости их следует очищать отдельно.
Для связанных служб, использующих Self-Hosted Integration Runtime, конфиденциальные сведения, такие как ключи или пароли, шифруются Self-Hosted Integration Runtime, чтобы гарантировать, что учетные данные остаются в пользовательской среде частной сети. Некоторые поля с конфиденциальными данными, на которые таким образом ссылается пользовательский код приложения, могут отсутствовать. При необходимости в extendedProperties используйте SecureString, а не ссылку на связанную службу.

Передача выходных данных в другое действие

Вы можете отправлять пользовательские значения из вашего кода в Пользовательской Деятельности обратно в службу. Вы можете сделать это, записав их в outputs.json из вашего приложения. Служба копирует содержимое файла outputs.json и добавляет его в выходные данные действия в качестве значения свойства customOutput. (Предел размера 2 МБ.) Если вы хотите использовать содержимое файла outputs.json в последующих действиях, значение можно получить с помощью выражения @activity('<MyCustomActivity>').output.customOutput.

Извлечение выходных данных SecureString

Конфиденциальные значения свойств, обозначенные как тип SecureString, как показано в некоторых примерах в этой статье, замаскированы на вкладке "Мониторинг" в пользовательском интерфейсе. Однако при фактическом выполнении конвейера свойство SecureString сериализуется как JSON в файле activity.json в виде обычного текста. Например:

"extendedProperties": {
  "connectionString": {
    "type": "SecureString",
    "value": "aSampleSecureString"
  }
}

Сериализация на самом деле небезопасна и не предназначена для обеспечения безопасности. Цель состоит в том, чтобы указать службе на то, что необходимо скрыть это значение на вкладке "Мониторинг".

Чтобы получить доступ к свойствам типа SecureString из пользовательской активности, прочитайте файл activity.json, который расположен в той же папке, что и ваш EXE-файл, выполните десериализацию JSON, а затем получите доступ к свойству JSON (extendedProperties => [propertyName] => value).

Автоматическое масштабирование Azure Batch

Вы также можете создать пул Azure Batch с помощью функции autoscale. Например, можно создать пул Azure Batch с 0 выделенными виртуальными машинами и формулой автомасштабирования на основе количества ожидающих задач.

Приведенный здесь пример формулы обеспечивает следующее поведение: при создании пула он изначально содержит одну виртуальную машину. Метрика $PendingTasks определяет количество задач в состоянии выполнения и активном состоянии (в очереди). Формула находит среднее число ожидающих выполнения задач за последние 180 секунд и соответствующим образом задает значение TargetDedicated. Благодаря этому значение TargetDedicated никогда не превысит 25 виртуальных машин. Таким образом, по мере поступления новых задач пул автоматически увеличивается, а по мере их завершения виртуальные машины высвобождаются по одной, и функция автоматического масштабирования уменьшает пул. Значения startingNumberOfVMs и maxNumberofVMs можно настроить в соответствии со своими потребностями.

Формула автоматического масштабирования:

startingNumberOfVMs = 1;
maxNumberofVMs = 25;
pendingTaskSamplePercent = $PendingTasks.GetSamplePercent(180 * TimeInterval_Second);
pendingTaskSamples = pendingTaskSamplePercent < 70 ? startingNumberOfVMs : avg($PendingTasks.GetSample(180 * TimeInterval_Second));
$TargetDedicated=min(maxNumberofVMs,pendingTaskSamples);

Дополнительные сведения см. в статье Автоматическое масштабирование вычислительных узлов в пуле Azure Batch.

Если в пуле используется autoScaleEvaluationInterval(значение по умолчанию), пакетной службе может потребоваться 15–30 минут на подготовку виртуальной машины перед выполнением настраиваемого действия. Если пул использует другое значение autoScaleEvaluationInterval, пакетная служба может затрачивать autoScaleEvaluationInterval плюс 10 минут.

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

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-04-21

Использование настраиваемых активностей в рамках конвейера Azure Data Factory или Azure Synapse Analytics

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

связанная служба Azure Batch

Настраиваемое действие

Разрешения настраиваемых действий

Выполнение команд

Передача объектов и свойств

Извлечение выходных данных выполнения

Передача выходных данных в другое действие

Извлечение выходных данных SecureString

Автоматическое масштабирование Azure Batch

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

Обратная связь

Дополнительные ресурсы