Utiliser des activités personnalisées dans un pipeline Azure Data Factory ou Azure Synapse Analytics

S’APPLIQUE À : Azure Data Factory Azure Synapse Analytics

Vous pouvez utiliser deux types d’activités dans un pipeline Azure Data Factory ou Synapse.

• Les activités de déplacement de données permettent de transférer des données entre les magasins de données source et récepteur pris en charge.
• Activités de transformation de données pour transformer les données à l’aide de services de calcul tels qu’Azure HDInsight et Azure Batch.

Pour déplacer des données vers ou depuis un magasin de données que le service ne prend pas en charge, ou pour transformer et traiter les données d’une manière qui n’est pas prise en charge par le service, créez une Activité personnalisée avec votre propre logique de déplacement ou de transformation des données, et utilisez cette activité dans un pipeline. L’activité personnalisée exécute votre logique de code personnalisé sur un pool de machines virtuelles Azure Batch.

Consultez les articles suivants si vous ne connaissez pas le service Azure Batch :

• Présentation de base d’Azure Batch pour une vue d’ensemble du service Azure Batch.
• Applet de commande New-AzBatchAccount pour créer un compte Azure Batch (ou) Portail Azure pour créer le compte Azure Batch à l’aide du portail Azure. Pour obtenir des instructions détaillées sur l’utilisation de la cmdlet, voir Utilisation de PowerShell pour gérer un compte Azure Batch.
• Applet de commande New-AzBatchPool pour créer un pool Azure Batch.

Ajouter des activités personnalisées à un pipeline avec l’interface utilisateur

Pour utiliser une activité Custom dans un pipeline, effectuez les étapes suivantes :

1. Recherchez Custom dans le volet Activités de pipeline, puis faites glisser une activité Custom vers le canevas du pipeline.

2. Sélectionnez la nouvelle activité personnalisée sur le canevas si elle n’est pas déjà sélectionnée.

3. Sélectionnez l’onglet Azure Batch pour sélectionner ou créer un service lié Azure Batch qui va exécuter l’activité personnalisée.

   

4. Sélectionnez l’onglet Paramètres et spécifiez une commande à exécuter sur Azure Batch, et des détails avancés facultatifs.

   

Service lié Azure Batch

L’extrait de code JSON suivant définit un exemple de service lié Azure Batch. Pour plus d’informations, consultez Environnements Compute pris en charge

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

Pour approfondir votre connaissance du service lié Azure Batch, consultez l’article Services liés de calcul.

Activité personnalisée

L’extrait de code JSON suivant définit un pipeline avec une activité personnalisée simple. La définition de l’activité comporte une référence au service lié 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"
        }
      }
    }]
  }
}

Dans cet exemple, le fichier helloworld.exe est une application personnalisée qui est stockée dans le dossier customactv2/helloworld du compte Stockage Azure utilisé dans le resourceLinkedService. L’activité personnalisée soumet cette application personnalisée pour qu’elle soit exécutée sur Azure Batch. Vous pouvez remplacer la commande par n’importe quelle application par défaut pouvant être exécutée sur le système d’exploitation cible des nœuds du pool Azure Batch.

Le tableau suivant indique les noms et les descriptions des propriétés qui sont spécifiques à cette activité.

* Les propriétés resourceLinkedService et folderPath doivent être toutes deux spécifiées ou omises.

Autorisations d’activité personnalisée

L’activité personnalisée définit le compte d’utilisateur automatique Azure Batch sur Accès non-administrateur avec étendue de la tâche (spécification utilisateur automatique par défaut). Vous ne pouvez pas modifier le niveau d’autorisation du compte d’utilisateur automatique. Pour plus d’informations, voir Exécuter des tâches sous des comptes d’utilisateur dans Azure Batch | Comptes d’utilisateur automatique.

Exécution de commandes

Vous pouvez exécuter directement une commande à l’aide d’une activité personnalisée. Dans l’exemple suivant, la commande « echo hello world » est exécutée sur les nœuds de pool Azure Batch cibles, et la sortie est envoyée vers 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"
      }
    }]
  }
}

Passage des objets et des propriétés

Cet exemple montre comment vous pouvez utiliser les referenceObjects et les extendedProperties pour passer des objets et des propriétés définies par l’utilisateur du service vers votre application personnalisée.

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

Lorsque l’activité est exécutée, les éléments referenceObjects et extendedProperties sont stockés dans les fichiers suivants qui sont déployés vers le même dossier d’exécution de l’application SampleApp.exe :

• activity.json

  Stocke extendedProperties et les propriétés de l’activité personnalisée.

• linkedServices.json

  Stocke un tableau des services liés définis dans la propriété referenceObjects.

• datasets.json

  Stocke un tableau de jeux de données défini dans la propriété referenceObjects.

L’exemple de code suivant montre comment SampleApp.exe peut accéder aux informations nécessaires à partir de fichiers 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);
        }
    }
}

Récupérer les sorties de l’exécution

Vous pouvez démarrer une exécution de pipeline à l’aide de la commande PowerShell suivante :

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

Quand le pipeline s’exécute, vous pouvez vérifier la sortie de l’exécution à l’aide des commandes suivantes :

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"

Les éléments stdout et stderr de votre application personnalisée sont enregistrés dans le conteneur adfjobs du service lié Stockage Azure que vous avez défini lors de la création du service lié Azure Batch à l’aide d’un GUID de la tâche. Vous pouvez obtenir le chemin détaillé à partir de la sortie de l’exécution de l’activité, comme indiqué dans l’extrait de code suivant :

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"

Si vous souhaitez consommer le contenu de stdout.txt dans des activités en aval, vous pouvez obtenir le chemin du fichier stdout.txt dans l’expression « @activity('MyCustomActivity').output.outputs[0] ».

Passer les sorties à une autre activité

Vous pouvez renvoyer les valeurs personnalisées figurant dans le code d’une activité personnalisée au service. Pour cela, vous devez les écrire dans le fichier outputs.json de votre application. Le service copie le contenu de outputs.json et l’ajoute à la sortie d’activité comme valeur de la propriété customOutput. (La taille limite est de 2 Mo.) Si vous souhaitez utiliser le contenu de outputs.json dans des activités en aval, vous pouvez obtenir la valeur à l’aide de l’expression @activity('<MyCustomActivity>').output.customOutput.

Récupérer les sorties SecureString

Les valeurs de propriété sensibles désignées en tant que type SecureString, comme illustré dans certains exemples de cet article, sont masquées dans l’onglet Surveillance de l’interface utilisateur. Lors de l’exécution réelle du pipeline, cependant, une propriété SecureString est sérialisée au format JSON dans le fichier activity.json en tant que texte brut. Par exemple :

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

Cette sérialisation n’est pas véritablement sécurisée et n’est pas destinée à être sécurisée. L’objectif est d’indiquer au service de masquer la valeur dans l’onglet Analyse.

Pour accéder aux propriétés de type SecureString à partir d’une activité personnalisée, lisez le fichier activity.json, placé dans le même dossier que le fichier EXE, désérialisez le code JSON, puis accédez à la propriété JSON (extendedProperties => [propertyName] => valeur).

Mise à l’échelle automatique d’Azure Batch

Vous pouvez aussi créer un pool Azure Batch avec la fonctionnalité autoscale . Par exemple, vous pouvez créer un pool Azure Batch avec 0 machine virtuelle dédiée et une formule de mise à l’échelle automatique en fonction du nombre de tâches en attente.

L’exemple de formule fourni ici entraîne le comportement suivant : Lors de sa création, le pool ne contient qu’une seule machine virtuelle. La métrique $PendingTasks définit le nombre de tâches dans l’état En cours d’exécution + Actif (en file d’attente). Cette formule recherche le nombre moyen de tâches en attente au cours des 180 dernières secondes et définit TargetDedicated en conséquence. Elle garantit que TargetDedicated ne va jamais au-delà de 25 machines virtuelles. Par conséquent, à mesure que de nouvelles tâches sont envoyées, le pool s’accroît automatiquement et, au fil de la réalisation des tâches, les machines virtuelles se libèrent une à une et la mise à l’échelle automatique réduit ces machines virtuelles. Vous pouvez ajuster startingNumberOfVMs et maxNumberofVMs selon vos besoins.

Formule de mise à l’échelle automatique :

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

Pour plus d’informations, consultez Mettre automatiquement à l’échelle les nœuds de calcul dans un pool Azure Batch .

Si le pool utilise la valeur par défaut du paramètre autoScaleEvaluationInterval, le service Batch peut mettre 15 à 30 minutes à préparer la machine virtuelle avant d’exécuter l’activité personnalisée. Si le pool utilise une autre valeur pour autoScaleEvaluationInterval, le service Batch peut prendre la durée d’autoScaleEvaluationInterval + 10 minutes.

Contenu connexe

Consultez les articles suivants qui expliquent comment transformer des données par d’autres moyens :

• Activité U-SQL
• Activité Hive
• Activité pig
• Activité MapReduce
• Activité de diffusion en continu Hadoop
• Activité Spark
• Activité de procédure stockée