Verwenden von benutzerdefinierten Aktivitäten in einer Azure Data Factory- oder Azure Synapse Analytics Pipeline

Gilt für: Azure Data Factory Azure Synapse Analytics

Es gibt zwei Arten von Aktivitäten, die Sie in einer Azure Data Factory- oder Synapse-Pipeline verwenden können.

• Datenverschiebungsaktivitäten zum Verschieben von Daten zwischen unterstützten Quell- und Senkendatenspeichern.
• Data-Transformationsaktivitäten, um Daten mithilfe von Rechendiensten wie Azure HDInsight und Azure Batch zu transformieren.

Wenn Sie Daten in einen bzw. aus einem von dem Dienst nicht unterstützten Datenspeicher verschieben oder auf eine Weise transformieren/verarbeiten müssen, die von dem Dienst nicht unterstützt wird, können Sie auch eine benutzerdefinierte Aktivität mit Ihrer eigenen Datenverschiebungs- bzw. -transformationslogik erstellen und in einer Pipeline verwenden. Die benutzerdefinierte Aktivität führt Ihre benutzerdefinierte Codelogik auf einem Azure Batch Pool virtueller Computer aus.

Lesen Sie die folgenden Artikel, wenn Sie neu beim Azure Batch-Dienst sind:

• Azure Batch Basics sind eine Übersicht über den Azure Batch-Dienst.
• New-AzBatchAccount Cmdlet zum Erstellen eines Azure Batch Kontos (oder) Azure Portal zum Erstellen des Azure Batch Kontos mithilfe Azure Portals. Ausführliche Anweisungen zur Verwendung des Cmdlets finden Sie unter Using PowerShell zum Verwalten von Azure Batch Account Artikel.
• New-AzBatchPool Cmdlet zum Erstellen eines Azure Batch Pools.

Hinzufügen benutzerdefinierter Aktivitäten zu einer Pipeline über die Benutzeroberfläche

Führen Sie die folgenden Schritte aus, um eine benutzerdefinierte Aktivität in einer Pipeline zu verwenden:

1. Suchen Sie im Bereich „Aktivitäten“ der Pipeline nach Benutzerdefiniert, und ziehen Sie eine benutzerdefinierte Aktivität in die Pipelinecanvas.

2. Wählen Sie die neue benutzerdefinierte Aktivität in der Canvas aus, wenn sie noch nicht ausgewählt ist.

3. Wählen Sie die Registerkarte Azure Batch aus, um einen neuen Azure Batch verknüpften Dienst auszuwählen oder zu erstellen, der die benutzerdefinierte Aktivität ausführt.

   

4. Wählen Sie die Registerkarte Settings aus, und geben Sie einen Befehl an, der auf dem Azure Batch ausgeführt werden soll, und optional erweiterte Details.

   

Azure Batch verbundener Dienst

Der folgende JSON-Code definiert ein Beispiel Azure Batch verknüpften Diensts. Weitere Informationen finden Sie unter Unterstützte Compute-Umgebungen

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

Weitere Informationen zum verknüpften Dienst Azure Batch finden Sie im Artikel Compute linked services.

Benutzerdefinierte Aktivität

Der folgende JSON-Ausschnitt definiert eine Pipeline mit einer einfachen benutzerdefinierten Aktivität. Die Aktivitätsdefinition enthält einen Verweis auf den verknüpften Azure Batch-Dienst.

{
  "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 diesem Beispiel ist die helloworld.exe eine benutzerdefinierte Anwendung, die im Ordner "customactv2/helloworld" des im resourceLinkedService verwendeten Azure Storage Kontos gespeichert ist. Die benutzerdefinierte Aktivität übermittelt diese benutzerdefinierte Anwendung, um sie auf Azure Batch auszuführen. Sie können den Befehl durch eine beliebige bevorzugte Anwendung ersetzen, die auf dem Zielbetriebssystem der Azure Batch Pool-Nodes ausgeführt werden kann.

Die folgende Tabelle beschreibt die Namen und Eigenschaften, die für diese Aktivität spezifisch sind.

* Die Eigenschaften resourceLinkedService und folderPath müssen entweder beide angegeben oder beide weggelassen werden.

Berechtigungen für benutzerdefinierte Aktivitäten

Die benutzerdefinierte Aktivität legt das Azure Batch automatischen Benutzerkonto auf Non-admin-Zugriff mit Aufgabenbereich (die Standardspezifikation für automatische Benutzer) fest. Sie können die Berechtigungsebene für das automatische Benutzerkonto nicht ändern. Weitere Informationen finden Sie unter Ausführen von Aufgaben unter Benutzerkonten in Batch | Automatische Benutzerkonten.

Ausführen von Befehlen

Mithilfe einer benutzerdefinierten Aktivität können Sie direkt einen Befehl ausführen. Im folgenden Beispiel wird der „Echo Hello World“-Befehl auf dem Azure Batch-Poolzielknoten ausgeführt und in „stdout“ ausgegeben.

{
  "name": "MyCustomActivity",
  "properties": {
    "description": "Custom activity sample",
    "activities": [{
      "type": "Custom",
      "name": "MyCustomActivity",
      "linkedServiceName": {
        "referenceName": "AzureBatchLinkedService",
        "type": "LinkedServiceReference"
      },
      "typeProperties": {
        "command": "cmd /c echo hello world"
      }
    }]
  }
}

Übergeben von Objekten und Eigenschaften

Dieses Beispiel zeigt, wie Sie das „referenceObjects“ und das „extendedProperties“ verwenden können, um Objekte und benutzerdefinierte Eigenschaften von dem Dienst an Ihre benutzerdefinierte Anwendung zu übergeben.

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

Wenn die Aktivität ausgeführt wird, werden „referenceObjects“ und „extendedProperties“ in folgenden Dateien gespeichert, die im selben Ausführungsordner von „SampleApp.exe“ bereitgestellt werden:

• activity.json

  Speichert erweiterte Eigenschaften und die Eigenschaften der benutzerdefinierten Aktivität.

• linkedServices.json

  Speichert ein Array verknüpfter Dienste, die in der „referenceObjects“-Eigenschaft definiert sind.

• datasets.json

  Speichert ein Array von Datasets, die in der „referenceObjects“-Eigenschaft definiert sind.

Der folgende Beispielcode veranschaulicht, wie „SampleApp.exe“ auf die erforderlichen Informationen in JSON-Dateien zugreifen kann:

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);
        }
    }
}

Ausführungsergebnisse abrufen

Sie können mit dem folgenden PowerShell-Befehl eine Pipelineausführung starten:

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

Wenn die Pipeline ausgeführt wird, können Sie die Ausgaben mit den folgenden Befehlen überprüfen:

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"

Die stdout und stderr Ihrer benutzerdefinierten Anwendung werden im adfjobs-Container des im Azure Storage verknüpften Dienstes gespeichert, den Sie beim Erstellen des Azure Batch-Linked Service mit einer GUID der Aufgabe definiert haben. Den detaillierten Pfad finden Sie in der Ausgabe der Aktivitätsausführung, wie im folgenden Ausschnitt gezeigt:

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"

Wenn Sie den Inhalt von „stdout.txt“ in Downstreamaktivitäten nutzen möchten, können Sie den Pfad zur Datei „stdout.txt“ im Ausdruck „@activity('MyCustomActivity').output.outputs[0]“ abrufen.

Übergeben von Ergebnissen an eine andere Aktivität

Sie können benutzerdefinierte Werte aus Ihrem Code in einer benutzerdefinierten Aktivität an den Dienst zurückgeben. Schreiben Sie sie hierzu in Ihrer Anwendung in outputs.json. Der Dienst kopiert den Inhalt von outputs.json und fügt ihn als Wert der Eigenschaft customOutput an die Aktivitätsausgabe an. (Die maximale Größe beträgt 2 MB.) Wenn Sie den Inhalt von outputs.json in Downstreamaktivitäten nutzen möchten, können Sie den Wert unter Verwendung des Ausdrucks @activity('<MyCustomActivity>').output.customOutput abrufen.

Abrufen von SecureString-Ausgaben

Vertrauliche Eigenschaftswerte, die als Typ SecureString definiert sind (wie in einigen der Beispiele in diesem Artikel gezeigt), werden auf der Registerkarte „Überwachung“ in der Benutzeroberfläche ausgeblendet. Bei der tatsächlichen Ausführung der Pipeline wird jedoch eine SecureString-Eigenschaft als JSON innerhalb der activity.json-Datei als Klartext serialisiert. Beispiel:

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

Diese Serialisierung ist nicht wirklich sicher und soll auch nicht sicher sein. Ziel ist es, dem Dienst einen Hinweis zu geben, den Wert auf der Registerkarte "Überwachung" zu maskieren.

Wenn Sie auf Eigenschaften des Typs SecureString von einer benutzerdefinierten Aktivität aus zugreifen möchten, lesen Sie die Datei activity.json, die sich in demselben Ordner wie Ihre EXE-Datei befindet, deserialisieren Sie den JSON-Code, und greifen Sie dann auf die JSON-Eigenschaft zu (extendedProperties => [propertyName] => Wert).

Automatische Skalierung von Azure Batch

Sie können auch einen Azure Batch Pool mit autoscale-Feature erstellen. Sie können z. B. einen Azure Batchpool mit 0 dedizierten VMs und einer Autoskalenformel basierend auf der Anzahl der ausstehenden Vorgänge erstellen.

Mit dieser Beispielformel wird folgendes Verhalten erreicht: Nachdem der Pool erstellt wurde, wird er mit einer VM gestartet. Die Metrik „$PendingTasks“ legt die Anzahl der Aufgaben im ausgeführten und im aktiven (in der Warteschlange) Zustand fest. Die Formel sucht nach der durchschnittlichen Anzahl ausstehender Aufgaben in den letzten 180 Sekunden und legt TargetDedicated auf den entsprechenden Wert fest. Dadurch wird sichergestellt, dass TargetDedicated nie die Anzahl von 25 virtuellen Computern überschreitet. Wenn also neue Aufgaben gesendet werden, wächst der Pool automatisch an. Beim Abschluss von Aufgaben werden virtuelle Computer nacheinander frei, und durch die automatische Skalierung werden diese virtuellen Computer reduziert. startingNumberOfVMs und maxNumberofVMs können entsprechend den jeweiligen Anforderungen angepasst werden.

Formel für die automatische Skalierung:

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

Weitere Informationen finden Sie unter Automatisch skalieren Computeknoten in einem Azure Batch Pool.

Wenn der Pool die Standardeinstellung für autoScaleEvaluationIntervalverwendet, kann es 15 bis 30 Minuten dauern, bis der Batch-Dienst den virtuellen Computer vorbereitet hat und die benutzerdefinierte Aktivität ausgeführt wird. Wenn der Pool eine andere Einstellung für „autoScaleEvaluationInterval“ nutzt, könnte der Batch-Dienst „autoScaleEvaluationInterval“ + 10 Minuten verwenden.

Zugehöriger Inhalt

In den folgenden Artikeln erfahren Sie, wie Daten auf andere Weisen transformiert werden:

• U-SQL-Aktivität
• Hive-Aktivität
• Pig-Aktivität
• MapReduce-Aktivität
• Hadoop-Streamingaktivität
• Spark-Aktivität
• Aktivität „Gespeicherte Prozedur“