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 armazenamentos de dados de origem e coletor 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 sua lógica de código personalizada em um pool de Lotes do Azure 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 Lote do Azure (ou) portal do Azure para criar a conta do Lote do Azure usando o portal do Azure. 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 de Lotes do Azure.

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 Personalizado no painel 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 a guia Lote do Azure para selecionar ou criar um novo serviço vinculado do Lote do Azure que executará a atividade personalizada.

   

4. Selecione a guia Configurações e especifique um comando a ser executado no Lote do Azure e detalhes avançados opcionais.

   

Serviço vinculado 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 vinculado do Lote do 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"
        }
      }
    }]
  }
}

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 envia este aplicativo personalizado para ser executado no Lote do Azure. Você pode substituir o comando para qualquer aplicativo preferencial que possa ser executado no Sistema Operacional de destino dos nós do Pool de Lotes 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 Lote do Azure como acesso não administrador com escopo de tarefa (a especificação de usuário automático padrão). 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 usuário no Batch | Contas de usuário 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 extendedProperties e 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 do seu aplicativo personalizado são salvos no contêiner adfjobs no Serviço Vinculado de Armazenamento do Azure que você definiu ao criar o Serviço Vinculado 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 você gostaria de consumir o conteúdo de stdout.txt em atividades downstream, você pode obter o caminho para o arquivo 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 em uma Atividade personalizada de volta para o serviço. Pode fazê-lo escrevendo-os a outputs.json partir da sua aplicação. O serviço copia o conteúdo e outputs.json o acrescenta à saída da atividade como o valor da customOutput propriedade. (O limite de tamanho é de 2MB.) Se você quiser consumir o conteúdo de outputs.json atividades a jusante, você pode obter o valor usando a expressão @activity('<MyCustomActivity>').output.customOutput.

Recuperar saídas SecureString

Os valores de propriedade confidenciais designados como tipo SecureString, conforme mostrado em alguns dos exemplos deste artigo, são mascarados na guia Monitoramento na interface do usuário. Na execução real do pipeline, no entanto, uma propriedade SecureString é serializada como JSON dentro do activity.json arquivo como texto sem formatação. 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 é uma dica para o serviço mascarar o valor na guia Monitoramento.

Para acessar propriedades do tipo SecureString a partir de uma atividade personalizada, leia o activity.json arquivo, que é colocado na mesma pasta que seu .EXE, desserialize o JSON e acesse a 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. $PendingTasks métrica define o número de tarefas em execução + estado ativo (enfileirado). 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 de faísca
• Atividade de procedimento armazenado