Использование пользовательских активностей в Фабрике данных Azure или конвейере Azure Synapse Analytics

ОБЛАСТЬ ПРИМЕНЕНИЯ: Фабрика данных Azure Azure Synapse Analytics

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

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

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

Если вы еще не знакомы с пакетной службой Azure, см. следующие статьи.

• Основные сведения о пакетной службе Azure — общие сведения о пакетной службе Azure.
• Вы можете создать учетную запись Azure Batch, используя командлет New-AzBatchAccount, либо воспользоваться порталом Azure для создания учетной записи через портал. Подробные инструкции по использованию этого командлета см. в статье Using PowerShell to manage Azure Batch Account (Использование PowerShell для управления учетной записью пакетной службы Azure).
• New-AzBatchPool cmdlet для создания пула пакетной службы Azure.

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

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

1. Найдите элемент Настраиваемое на панели конвейера "Действия" и перетащите его на холст конвейера.

2. Выберите настраиваемое действие на холсте, если оно еще не выбрано.

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

   

4. Откройте вкладку Параметры и укажите команду для выполнения в Пакетной службе Azure, а также дополнительные сведения.

   

Связанный пакетный сервис Azure

Ниже приведен фрагмент кода JSON, который определяет пример связанной пакетной службы Azure. Подробнее см. Поддерживаемые вычислительные среды

{
    "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 см. в статье Связанные вычислительные службы.

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

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

{
  "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, используемой в resourceLinkedService. Пользовательская активность отправляет это пользовательское приложение на выполнение в Azure Batch. Вы можете заменить команду на любое предпочитаемое приложение, которое можно выполнить в целевой операционной системе узлов пула пакетной службы Azure.

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

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

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

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

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

Вы можете непосредственно выполнить команду, используя настраиваемое действие. В следующем примере выполняется команда "echo hello world" на целевых узлах пула Azure Batch Pool и выводит результат в 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, определенной при создании связанной пакетной службы Azure с помощью 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]".

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

Вы можете отправлять пользовательские значения из вашего кода в Пользовательской Деятельности обратно в службу. Вы можете сделать это, записав их в 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

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

Приведенный здесь пример формулы обеспечивает следующее поведение: при создании пула он изначально содержит одну виртуальную машину. Метрика $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.

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

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

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

• Действие U-SQL
• Активность Hive
• Действие Pig
• Действие MapReduce
• Активность Hadoop Streaming
• Активность Spark
• Активность хранимой процедуры