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 do Synapse.

• Atividades de movimentação de dados para mover dados entre a fonte suportada e o armazenamento de dados de coletor.
• Atividades de transformação de dados para transformar dados usando serviços de computação, como Azure HDInsight e Lote do Azure.

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

Consulte os seguintes artigos caso seja novo no serviço Lote do Azure:

• Lote do Azure noções básicas para obter uma visão geral do serviço Lote do Azure.
• New-AzBatchAccount cmdlet para criar uma conta Lote do Azure (ou) Azure portal para criar a conta Lote do Azure usando Azure portal. Consulte O uso do PowerShell para gerenciar Lote do Azure Conta artigo para obter instruções detalhadas sobre como usar o cmdlet.
• New-AzBatchPool cmdlet para criar um pool de Lote do Azure.

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

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

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

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

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

   

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

   

Serviço vinculado do Lote do Azure

O JSON a seguir define um exemplo de serviço vinculado do Lote do Azure. Para saber mais, veja Ambientes de computação com suporte

{
    "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 vinculado do Lote do Azure, consulte o artigo Compute linked services.

Atividade personalizada

O snippet de código JSON a seguir define um pipeline com uma atividade personalizada simples. A definição de atividade tem uma referência ao serviço vinculado 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 Armazenamento do Azure usada no resourceLinkedService. A atividade Personalizada envia esse aplicativo personalizado a 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 Lote do Azure.

A tabela a seguir descreve os nomes e as descrições de propriedades que são específicas a esta atividade.

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

Permissões de atividade personalizada

A atividade personalizada define a conta de usuário automático do Lote do Azure como Acesso não-admin com escopo de tarefa (a especificação padrão do usuário automático). Você não pode alterar o nível de permissão da conta de usuário automático. Para obter mais informações, veja Executar tarefas em contas de usuário no Lote | Contas de usuário automático.

Executando comandos

É possível executar diretamente um comando usando a Atividade Personalizada. O exemplo a seguir executa o comando "eco Olá, mundo" nos nós de destino do Pool do Lote do Azure 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 as propriedades da atividade personalizada.

• linkedServices.json

  Armazena uma matriz de serviços vinculados definidos na propriedade referenceObjects.

• datasets.json

  Armazena uma matriz de conjuntos de dados definidos na propriedade referenceObjects.

O exemplo de código a seguir demonstra como o SampleApp.exe pode acessar as informações necessárias dos 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 as saídas de execução

É possível iniciar uma execução de pipeline, usando este comando do PowerShell:

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

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

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 stderr do aplicativo personalizado são salvos no contêiner adfjobs no serviço vinculado do Armazenamento do Microsoft Azure que você definiu durante a criação do serviço vinculado do Lote do Azure com um GUID da tarefa. Você pode obter o caminho detalhado da saída de execução da atividade, conforme mostrado no trecho de código a seguir:

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 deseja consumir o conteúdo de stdout.txt nas 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. Você pode fazer isso escrevendo-os no outputs.json a partir do seu aplicativo. O serviço copia o conteúdo do outputs.json e acrescenta-o para a Saída da Atividade como o valor da propriedade customOutput. O limite de tamanho é 2 MB. Se você quiser consumir o conteúdo de outputs.json em atividades downstream, você poderá obter o valor usando a expressão @activity('<MyCustomActivity>').output.customOutput.

Recuperar saídas de SecureString

Os valores de propriedades 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 no arquivo activity.json como texto simples. Por exemplo:

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

Essa serialização não é verdadeiramente segura e não é destinada a ser segura. A intenção é uma indicação ao serviço para mascarar o valor na aba Monitoramento.

Para acessar propriedades do tipo SecureString de uma atividade personalizada, leia o arquivo activity.json, que é colocado na mesma pasta que o .EXE, desserialize o JSON e, em seguida, acesse a propriedade JSON (extendedProperties => [propertyName] => value).

Dimensionamento automático do Lote do Azure

Você também pode criar um pool de Lote do Azure com o recurso autoscale. Por exemplo, você pode criar um pool de batch do Azure com 0 VMs dedicadas e uma fórmula de autoscalação com base no número de tarefas pendentes.

A fórmula de exemplo alcança o seguinte comportamento: quando o pool é criado pela primeira vez, ele começa com uma VM. A métrica de $PendingTasks define o número de tarefas em execução + estado ativo (em fila). A fórmula localiza o número médio de tarefas pendentes nos últimos 180 segundos e define TargetDedicated adequadamente. Isso garante que TargetDedicated nunca ultrapasse 25 VMs. Assim, o pool aumenta automaticamente conforme novas tarefas são enviadas e, conforme as tarefas são concluídas, as VMs se liberam uma a uma e são reduzidas pelo dimensionamento automático. startingNumberOfVMs e maxNumberofVMs podem ser ajustados às suas necessidades.

Fórmula de dimensionamento automático:

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 A escala automática de nós de computação em um pool de Lote do Azure para obter detalhes.

Se o pool estiver usando o padrão autoScaleEvaluationInterval, o serviço Lote poderá demorar 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 de lote pode levar autoScaleEvaluationInterval + 10 minutos.

Conteúdo relacionado

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

• U-SQL activity (Atividade do U-SQL)
• Atividade do Hive
• Atividade do Pig
• MapReduce activity (Atividade do MapReduce)
• Hadoop Streaming activity (Atividade de streaming do Hadoop)
• Spark activity (Atividade do Spark)
• Stored procedure activity (Atividade de procedimento armazenado)