Usar atividades personalizadas em um pipeline do Azure Data Factory ou do Azure Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Há dois tipos de atividades que você pode usar em um pipeline do Azure Data Factory ou Synapse.

• Atividades de movimentação de dados para mover dados entre repositórios de dados de origem e destinos suportados.
• Atividades de transformação de dados para transformar dados usando serviços de computação como o Azure HDInsight e o Azure Batch.

Para mover dados de/para um armazenamento de dados que o serviço não oferece suporte ou para transformar/processar dados de uma forma que não seja suportada pelo serviço, você pode criar uma atividade personalizada com sua própria lógica de movimentação ou transformação de dados e usar a atividade em um pipeline. A atividade personalizada executa a sua lógica de código personalizado num pool do Azure Batch de máquinas virtuais.

Consulte os seguintes artigos se você for novo no serviço Batch do Azure:

• Noções básicas do Azure Batch para obter uma visão geral do serviço Azure Batch.
• Cmdlet New-AzBatchAccount para criar uma conta de Batch do Azure ou portal do Azure para criar a conta de Batch do Azure através do portal. Consulte o artigo Usando o PowerShell para gerenciar a Conta em Lote do Azure para obter instruções detalhadas sobre como usar o cmdlet.
• Cmdlet New-AzBatchPool para criar um pool do Azure Batch.

Adicionar atividades personalizadas a um pipeline com a interface do usuário

Para usar uma atividade personalizada em um pipeline, conclua as seguintes etapas:

1. Procure por Personalizado no painel de Atividades do pipeline e arraste uma atividade Personalizada para a tela do pipeline.

2. Selecione a nova atividade Personalizada na tela, se ainda não estiver selecionada.

3. Selecione o separador Azure Lote para selecionar ou criar um novo serviço ligado do Azure Lote que executará a atividade personalizada.

   

4. Selecione o separador Configurações e especifique um comando para ser executado no Azure Batch, bem como detalhes avançados opcionais.

   

Serviço associado do Azure Batch

O JSON a seguir define um exemplo de serviço vinculado do Azure Batch. Para obter detalhes, consulte Ambientes de computação suportados

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

Para saber mais sobre o serviço vinculado do Azure Batch, consulte o artigo Serviços vinculados de computação.

Atividade personalizada

O trecho JSON a seguir define um pipeline com uma Atividade Personalizada simples. A definição de atividade tem uma referência ao serviço associado 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"
        }
      }
    }]
  }
}

Neste exemplo, o helloworld.exe é um aplicativo personalizado armazenado na pasta customactv2/helloworld da conta de Armazenamento do Azure usada no resourceLinkedService. A atividade personalizada submete esta aplicação personalizada para ser executada no Azure Batch. Você pode substituir o comando por qualquer aplicativo preferido que possa ser executado no sistema operativo de destino dos nós da Batch Pool do Azure.

A tabela a seguir descreve nomes e descrições de propriedades específicas para essa atividade.

* As propriedades resourceLinkedService e folderPath devem ser especificadas ou ambas devem ser omitidas.

Permissões de atividade personalizadas

A atividade personalizada define a conta de usuário automático do Azure Batch como acesso de não administrador com escopo para tarefas (a especificação padrão de usuário automático). Não é possível alterar o nível de permissão da conta de usuário automático. Para obter mais informações, consulte Executar tarefas em contas de utilizador no Batch | Contas de utilizador automático.

Execução de comandos

Você pode executar diretamente um comando usando a Atividade personalizada. O exemplo a seguir executa o comando "echo hello world" nos nós do Pool de Lotes do Azure de destino e imprime a saída para 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"
      }
    }]
  }
}

Passando objetos e propriedades

Este exemplo mostra como você pode usar referenceObjects e extendedProperties para passar objetos e propriedades definidas pelo usuário do serviço para seu aplicativo personalizado.

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

Quando a atividade é executada, referenceObjects e extendedProperties são armazenados nos seguintes arquivos que são implantados na mesma pasta de execução do SampleApp.exe:

• activity.json

  Armazena as propriedades estendidas e as propriedades da atividade personalizada.

• linkedServices.json

  Armazena uma matriz de Serviços Vinculados definidos na propriedade referenceObjects.

• datasets.json

  Armazena uma matriz de Datasets definida na propriedade referenceObjects.

O código de exemplo a seguir demonstra como o SampleApp.exe pode acessar as informações necessárias de arquivos 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);
        }
    }
}

Recuperar saídas de execução

Você pode iniciar uma execução de pipeline usando o seguinte comando do PowerShell:

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

Quando o pipeline está em execução, você pode verificar a saída de execução usando os seguintes comandos:

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"

O stdout e o stderr da sua aplicação personalizada são guardados no contentor adfjobs no Serviço Ligado de Armazenamento do Azure que foi definido ao criar o Serviço Ligado de Processamento em Lote do Azure com um GUID da tarefa. Você pode obter o caminho detalhado da saída Activity Run, conforme mostrado no seguinte trecho:

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"

Se pretender consumir o conteúdo de stdout.txt em atividades subsequentes, pode obter o caminho para o ficheiro stdout.txt na expressão "@activity('MyCustomActivity').output.outputs[0]".

Passar saídas para outra atividade

Você pode enviar valores personalizados do seu código numa Atividade Personalizada de volta para o serviço. Pode fazê-lo escrevendo-os na outputs.json a partir da sua aplicação. O serviço copia o conteúdo de outputs.json e o acrescenta à saída da atividade como valor da propriedade customOutput. (O limite de tamanho é de 2MB.) Se pretendes consumir o conteúdo de outputs.json em atividades a jusante, podes obter o valor usando a expressão @activity('<MyCustomActivity>').output.customOutput.

Recuperar resultados SecureString

Os valores de propriedade confidenciais designados como tipo SecureString, conforme mostrado em alguns dos exemplos deste artigo, são mascarados na aba de Monitorização na interface de utilizador. Na execução real do pipeline, no entanto, uma propriedade SecureString é serializada como JSON dentro do activity.json arquivo como texto simples. Por exemplo:

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

Esta serialização não é verdadeiramente segura e não se destina a ser segura. A intenção é dar ao serviço a dica de mascarar o valor na guia Monitorização.

Para aceder a propriedades do tipo SecureString a partir de uma atividade personalizada, leia o ficheiro activity.json, que é colocado na mesma pasta que o seu ficheiro .EXE, desserializar o JSON e aceda à propriedade JSON (extendedProperties => [propertyName] => value).

Dimensionamento automático do Azure Batch

Você também pode criar um pool de lotes do Azure com o recurso de dimensionamento automático. Por exemplo, você pode criar um pool de lotes do Azure com 0 VMs dedicadas e uma fórmula de dimensionamento automático com base no número de tarefas pendentes.

A fórmula de exemplo aqui alcança o seguinte comportamento: Quando o pool é criado inicialmente, ele começa com 1 VM. A métrica $PendingTasks define o número de tarefas em estado de execução + ativo (em fila). A fórmula localiza o número médio de tarefas pendentes nos últimos 180 segundos e define TargetDedicated de acordo. Ele garante que o TargetDedicated nunca vá além de 25 VMs. Assim, à medida que novas tarefas são enviadas, o pool cresce automaticamente e, à medida que as tarefas são concluídas, as VMs ficam livres uma a uma e o dimensionamento automático reduz essas VMs. startingNumberOfVMs e maxNumberofVMs podem ser ajustados às suas necessidades.

Fórmula de escala automática:

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

Consulte Dimensionar automaticamente nós de computação em um pool de lotes do Azure para obter detalhes.

Se o pool estiver usando o autoScaleEvaluationInterval padrão, o serviço em lote poderá levar de 15 a 30 minutos para preparar a VM antes de executar a atividade personalizada. Se o pool estiver usando um autoScaleEvaluationInterval diferente, o serviço Batch poderá levar autoScaleEvaluationInterval + 10 minutos.

Conteúdos relacionados

Consulte os seguintes artigos que explicam como transformar dados de outras maneiras:

• Atividade U-SQL
• Atividade da colmeia
• Atividade suína
• Atividade MapReduce
• Atividade de streaming do Hadoop
• Atividade Spark
• Atividade de procedimento armazenado