Use atividades personalizadas num pipeline do Azure Data Factory ou Azure Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Existem dois tipos de atividades que pode usar num 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 Azure HDInsight e 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 personalizada num pool Azure Batch de máquinas virtuais.

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

• Noções básicas do Azure Batch para uma visão geral do serviço Azure Batch.
• New-AzBatchAccount cmdlet para criar uma conta de Azure Batch (ou) Azure portal para criar a conta Azure Batch usando Azure portal. Consulte o artigo Using PowerShell to manage Azure Batch Account para instruções detalhadas sobre como usar o cmdlet.
• New-AzBatchPool cmdlet para criar um pool de 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 Batch para selecionar ou criar um novo serviço Azure Batch ligado que execute a atividade personalizada.

   

4. Selecione o separador Settings e especifique um comando a executar no Azure Batch, além de detalhes avançados opcionais.

   

Azure Batch serviço associado

O JSON seguinte define um serviço ligado Azure Batch de exemplo. 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 mais informações sobre o serviço ligado do Azure Batch, consulte o artigo Compute linked services.

Atividade personalizada

O trecho JSON a seguir define um pipeline com uma Atividade Personalizada simples. A definição de atividade faz referência ao serviço ligado 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 é uma aplicação personalizada armazenada na pasta customactv2/helloworld da conta Armazenamento do Azure usada no resourceLinkedService. A atividade Personalizada submete esta aplicação personalizada para ser executada no Azure Batch. Pode substituir o comando por qualquer aplicação preferida que possa ser executada no Sistema Operativo alvo dos nós do Azure Batch Pool.

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 utilizador automático Azure Batch para Acesso não administrativo com âmbito de tarefa (a especificação padrão do utilizador 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 seguinte executa o comando "echo hello world" nos nós do Azure Batch Pool de destino e imprime a saída no 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"

Os stdout e stderr da sua aplicação personalizada são guardados no contentor adfjobs no Serviço Ligado do Armazenamento do Azure que definiu ao criar o Serviço Ligado do Azure Batch com o 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).

Autoescalonamento do Azure Batch

Também pode criar um pool de Azure Batch com a funcionalidade autoscale. Por exemplo, podes criar um pool de lotes no Azure com 0 VMs dedicadas e uma fórmula de autoescala baseada 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 Escalar automaticamente nós de computação num pool de Azure Batch para mais 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