Usare attività personalizzate in una pipeline di Azure Data Factory o di Azure Synapse Analytics

APPLICABILE A: Azure Data Factory Azure Synapse Analytics

Esistono due tipi di attività che è possibile usare in una pipeline Azure Data Factory o Synapse.

• Attività di spostamento dei dati per spostare i dati tra archivi dati di origine e di destinazione supportati.
• Attività di trasformazione Dati per trasformare i dati usando servizi di calcolo come Azure HDInsight e Azure Batch.

Per spostare i dati da e verso un archivio dati non supportato dal servizio o per trasformare/elaborare i dati in modo non supportato dal servizio, è possibile creare un'attività personalizzata con la propria logica di spostamento o trasformazione dei dati e usare l'attività in una pipeline. L'attività personalizzata esegue la logica del codice personalizzata in un pool di macchine virtuali Azure Batch.

Se si è nuovi al servizio Azure Batch, vedere gli articoli seguenti:

• Azure Batch nozioni di base per una panoramica del servizio Azure Batch.
• New-AzBatchAccount cmdlet per creare un account Azure Batch (o) Azure portal per creare l'account Azure Batch usando Azure portal. Per istruzioni dettagliate sull'uso del cmdlet, vedere l'articolo Using PowerShell to manage Azure Batch Account.
• cmdlet New-AzBatchPool per creare un pool di Azure Batch.

Aggiungere attività personalizzate a una pipeline con l'interfaccia utente

Per usare un'attività personalizzata in una pipeline, completare la procedura seguente:

1. Cercare Personalizza nel riquadro Attività della pipeline, quindi trascinare un'attività Personalizza nel canvas della pipeline.

2. Selezionare la nuova attività Personalizzata nell'area di disegno, se non è già selezionata.

3. Selezionare la scheda Azure Batch per selezionare o creare un nuovo servizio collegato Azure Batch che eseguirà l'attività personalizzata.

   

4. Selezionare la scheda Impostazioni e specificare un comando da eseguire nel Azure Batch e dettagli avanzati facoltativi.

   

Azure Batch servizio collegato

Il codice JSON seguente definisce un esempio Azure Batch servizio collegato. Per informazioni dettagliate, vedere Ambienti di calcolo supportati

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

Per ulteriori informazioni sul servizio collegato Azure Batch, consultare l'articolo Servizi collegati di calcolo.

Attività personalizzata

Il frammento di codice JSON seguente definisce una pipeline con una semplice attività personalizzata. La definizione dell'attività ha un riferimento al servizio collegato 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"
        }
      }
    }]
  }
}

In questo esempio, il helloworld.exe è un'applicazione personalizzata archiviata nella cartella customactv2/helloworld dell'account Archiviazione di Azure usato nella risorsaLinkedService. L'attività Personalizzata invia l'applicazione personalizzata da eseguire in Azure Batch. È possibile sostituire il comando con qualsiasi applicazione preferita che può essere eseguita nel sistema operativo di destinazione dei nodi pool Azure Batch.

Nella tabella seguente vengono descritti i nomi e le descrizioni delle proprietà specifiche per questa attività.

* Le proprietà resourceLinkedService e folderPath devono essere specificate o entrambe o entrambe omesse.

Autorizzazioni per le attività personalizzate

L'attività personalizzata imposta l'account utente automatico di Azure Batch su Utente non amministratore con ambito attività (specifica predefinita dell'utente automatico). Non è possibile modificare il livello di autorizzazione dell'account utente automatico. Per altre informazioni, vedere Eseguire attività con account utente in Batch | Account utente automatici.

Esecuzione di comandi

È possibile eseguire direttamente un comando tramite l'attività personalizzata. L'esempio seguente esegue il comando "echo hello world" sui nodi di destinazione del pool di Azure Batch e stampa l'output su 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"
      }
    }]
  }
}

Passaggio di oggetti e proprietà

Questo esempio illustra come usare referenceObjects e extendedProperties per passare oggetti e proprietà definite dall'utente dal servizio all'applicazione personalizzata.

{
  "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 l'attività viene eseguita, le proprietà referenceObjects ed extendedProperties vengono archiviate nei file descritti di seguito, che vengono distribuiti nella stessa cartella di esecuzione di SampleApp.exe:

• activity.json

  Contiene extendedProperties e le proprietà dell'attività personalizzata.

• linkedServices.json

  Contiene una matrice di servizi collegati definiti nella proprietà referenceObjects.

• datasets.json

  Contiene una matrice di set di dati definiti nella proprietà referenceObjects.

Il codice di esempio seguente illustra come SampleApp.exe è in grado di accedere alle informazioni necessarie dai file 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);
        }
    }
}

Recuperare i risultati di esecuzione

Per avviare una pipeline, eseguire il comando di PowerShell seguente:

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

Quando la pipeline è in esecuzione, è possibile controllare l'output dell'esecuzione usando i comandi seguenti:

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"

Il stdout e stderr dell'applicazione personalizzata vengono salvati nel contenitore adfjobs nel servizio collegato di Archiviazione di Azure che hai definito durante la creazione del servizio collegato di Azure Batch, con un GUID dell'attività. È possibile ottenere il percorso dettagliato dall'output di Esecuzione attività, come illustrato nel frammento di codice seguente:

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 si desidera utilizzare il contenuto di stdout.txt nelle attività downstream, è possibile ottenere il percorso del file stdout.txt nell'espressione "@activity('MyCustomActivity').outputs[0]".

Passare gli output a un'altra attività

È possibile inviare valori personalizzati dal codice di un'attività personalizzata al servizio. È possibile farlo scrivendoli in outputs.json dall'applicazione. Il servizio copia il contenuto di outputs.json e lo aggiunge all'Activity Output come valore della proprietà customOutput. Il limite di dimensioni è 2 MB. Se si desidera utilizzare il contenuto di outputs.json nelle attività downstream, è possibile ottenere il valore usando l'espressione @activity('<MyCustomActivity>').output.customOutput.

Recuperare gli output SecureString

I valori delle proprietà sensibili designati come tipo SecureString, come illustrato in alcuni esempi di questo articolo, vengono mascherati nella scheda Monitoraggio nell'interfaccia utente. Nell'esecuzione effettiva della pipeline, tuttavia, una proprietà SecureString viene serializzata come JSON all'interno del activity.json file come testo normale. Ad esempio:

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

Questa serializzazione non è propriamente sicura e non è progettata per esserlo. L'intenzione è un'indicazione per il servizio di mascherare il valore nella scheda Monitoraggio.

Per accedere alle proprietà di tipo SecureString da un'attività personalizzata, leggere il activity.json file, che viene inserito nella stessa cartella del .EXE, deserializzare il codice JSON e quindi accedere alla proprietà JSON (extendedProperties => [propertyName] => value).

Ridimensionamento automatico di Azure Batch

È anche possibile creare un pool di Azure Batch con autoscale. Ad esempio, è possibile creare un pool batch Azure con 0 macchine virtuali dedicate e una formula di scalabilità automatica in base al numero di attività in sospeso.

La formula di esempio seguente consente di ottenere il comportamento seguente: quando il pool viene creato inizialmente, inizia con 1 macchina virtuale. La metrica $PendingTasks definisce il numero di attività in esecuzione e quelle in coda. La formula trova il numero medio di attività in sospeso negli ultimi 180 secondi e imposta TargetDedicated di conseguenza. Assicura che TargetDedicated non vada mai oltre 25 macchine virtuali. Pertanto, quando vengono inviate nuove attività, il pool si espande automaticamente e al completamento delle attività le macchine virtuali diventano disponibili una alla volta e la scalabilità automatica le riduce. È possibile regolare startingNumberOfVMs e maxNumberofVMs in base alle esigenze.

Formula di scalabilità automatica:

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

Per informazioni dettagliate, vedere Scalabilità automatica dei nodi di calcolo in un pool di Azure Batch.

Se il pool usa il valore predefinito autoScaleEvaluationInterval, possono essere necessari 15-30 minuti perché il servizio Batch prepari la VM prima di eseguire l'attività personalizzata. Se il pool usa un valore autoScaleEvaluationInterval diverso, il servizio Batch può richiedere un valore autoScaleEvaluationInterval + 10 minuti.

Contenuto correlato

Vedere gli articoli seguenti, che illustrano altre modalità di trasformazione dei dati:

• Attività U-SQL
• Attività Hive
• Attività di Pig
• Attività MapReduce
• Attività di streaming di Hadoop
• Attività Spark
• Attività stored procedure