Uso de actividades personalizadas en una canalización de Azure Data Factory o Azure Synapse Analytics

Artículo
12/09/2023

SE APLICA A: Azure Data Factory Azure Synapse Analytics

Sugerencia

Pruebe Data Factory en Microsoft Fabric, una solución de análisis todo en uno para empresas. Microsoft Fabric abarca todo, desde el movimiento de datos hasta la ciencia de datos, el análisis en tiempo real, la inteligencia empresarial y los informes. Obtenga información sobre cómo iniciar una nueva evaluación gratuita.

Hay dos tipos de actividades que puede usar en una canalización de Azure Data Factory o de Synapse.

Actividades de movimiento de datos para mover datos entre almacenes de datos de origen y receptor compatibles.
Actividades de transformación de datos para transformar datos mediante servicios de proceso como Azure HDInsight y Azure Batch.

Para mover datos desde y hacia un almacén de datos incompatible con el servicio, o para transformar o procesar datos de algún modo incompatible con el servicio, puede crear una actividad personalizada con su propia lógica de movimiento o transformación de datos y usarla en una canalización. La actividad personalizada ejecuta la lógica del código personalizado en un grupo de máquinas virtuales de Azure Batch.

Nota

Se recomienda usar el módulo Azure Az de PowerShell para interactuar con Azure. Consulte Instalación de Azure PowerShell para empezar. Para más información sobre cómo migrar al módulo Az de PowerShell, consulte Migración de Azure PowerShell de AzureRM a Az.

Consulte los artículos siguientes si no está familiarizado con el servicio Azure Batch:

Aspectos básicos de Azure Batch para información general del servicio Azure Batch.
Cmdlet New-AzBatchAccount para crear una cuenta de Azure Batch, o Azure Portal para crear la cuenta de Azure Batch con Azure Portal. Consulte el artículo Using PowerShell to manage Azure Batch Account (Administración de cuentas de Azure Batch con PowerShell) para instrucciones detalladas sobre el uso del cmdlet.
New-AzBatchPool para crear un grupo de Azure Batch.

Importante

Al crear un grupo de Azure Batch nuevo, se debe usar "VirtualMachineConfiguration", NO "CloudServiceConfiguration". Para más información, consulte la guía de migración de grupos de Azure Batch.

Incorporación de actividades personalizadas a una canalización con la interfaz de usuario

Para usar una actividad personalizada en una canalización, complete los pasos siguientes:

Busque el valor Personalizado en el panel Actividades de canalización y arrastre una actividad personalizada al lienzo de canalización.
Seleccione la nueva actividad personalizada en el lienzo si aún no está seleccionada.
Seleccione la pestaña Azure Batch para seleccionar o crear un nuevo servicio vinculado de Azure Batch que ejecutará la actividad personalizada.
Seleccione la pestaña Configuración y especifique un comando para que se ejecute en Azure Batch, y detalles avanzados opcionales.

Servicio vinculado de Azure Batch

El siguiente JSON define un servicio vinculado de Azure Batch de ejemplo. Para obtener más información, consulte Entorno de proceso compatibles

{
    "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 más información sobre el servicio vinculado de Azure Batch, vea el artículo Servicios vinculados de proceso.

Actividad personalizada

El siguiente fragmento de código JSON define una canalización con una actividad personalizada simple. La definición de actividad tiene una referencia al servicio vinculado de 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"
        }
      }
    }]
  }
}

En este ejemplo, el archivo helloworld.exe es una aplicación personalizada que se almacena en la carpeta customactv2/helloworld de la cuenta de Azure Storage usada en la propiedad resourceLinkedService. La actividad personalizada envía esta aplicación personalizada para ejecutarla en Azure Batch. Puede reemplazar el comando en cualquier aplicación preferida que se pueda ejecutar en el sistema operativo de destino de los nodos del grupo de Azure Batch.

En la tabla siguiente se describen los nombres y descripciones de las propiedades que son específicas de esta actividad.

Propiedad	Descripción	Obligatorio
name	Nombre de la actividad en la canalización	Sí
description	Texto que describe para qué se usa la actividad.	No
type	Para la actividad personalizada, el tipo de actividad es Custom.	Sí
linkedServiceName	Servicio vinculado a Azure Batch. Para obtener más información sobre este servicio vinculado, vea el artículo Compute linked services (Servicios vinculados de procesos).	Sí
command	Comando de la aplicación personalizada que se va a ejecutar. Si la aplicación ya está disponible en el nodo del grupo de Azure Batch, se pueden omitir las propiedades resourceLinkedService y folderPath. Por ejemplo, puede especificar que el comando sea `cmd /c dir`, que el nodo del grupo de lotes de Windows admite de forma nativa.	Sí
resourceLinkedService	Servicio de Azure Storage vinculado a la cuenta de almacenamiento en la que está almacenada la aplicación personalizada	No *
folderPath	Ruta de acceso a la carpeta de la aplicación personalizada y todas sus dependencias Si tiene dependencias que se almacenan en subcarpetas (es decir, en una estructura jerárquica de carpetas bajo folderPath) la estructura de carpetas se elimina cuando los archivos se copian en Azure Batch. Es decir, todos los archivos se copian en una sola carpeta sin subcarpetas. Para evitar este comportamiento, considere la posibilidad de comprimir los archivos, copiar el archivo comprimido y, a continuación, descomprimirlo con código personalizado en la ubicación deseada.	No *
referenceObjects	Matriz de servicios vinculados y conjuntos de datos existentes. Los servicios vinculados y los conjuntos de datos a los que se hace referencia se pasan a la aplicación personalizada en formato JSON, por lo que el código personalizado puede hacer referencia a recursos del servicio	No
extendedProperties	Propiedades definidas por el usuario que se pueden pasar a la aplicación personalizada en formato JSON, por lo que el código personalizado puede hacer referencia a propiedades adicionales	No
retentionTimeInDays	Tiempo de retención de los archivos enviados para la actividad personalizada. El valor predeterminado es 30 días.	No

* Las propiedades resourceLinkedService y folderPath deben especificarse u omitirse de forma conjunta.

Nota

Si se pasan servicios vinculados como referenceObjects en la actividad personalizada, una buena práctica de seguridad consiste en pasar un servicio vinculado habilitado para Azure Key Vault (ya que no contiene cadenas seguras) y obtener las credenciales usando el nombre del secreto directamente de Key Vault desde el código. Aquí encontrará un ejemplo donde se hace referencia a un servicio vinculado habilitado para AKV, se recuperan las credenciales de Key Vault y, después, se accede al almacenamiento en el código.

Nota

Actualmente, solo Azure Blob Storage es compatible con resourceLinkedService en la actividad personalizada, y es el único servicio vinculado que se crea de manera predeterminada y no existe ninguna opción para elegir otros conectores como ADLS Gen2.

Permisos de la actividad personalizada

La actividad personalizada establece la cuenta de usuario automático de Azure Batch en Acceso sin privilegios de administrador con ámbito de la tarea (la especificación de usuario automático predeterminada). No se puede cambiar el nivel de permiso de la cuenta de usuario automático. Para más información, consulte Ejecución de tareas en cuentas de usuario en Batch | Cuentas de usuario automático.

Ejecutar comandos

Puede ejecutar directamente un comando mediante la actividad personalizada. En el ejemplo siguiente se ejecuta un comando "echo hello world" en los nodos de grupo de destino de Azure Batch y se imprime la salida en 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"
      }
    }]
  }
}

Pasar objetos y propiedades

En este ejemplo se muestra cómo usar las propiedades referenceObjects y extendedProperties para pasar objetos y propiedades definidas por el usuario del servicio a la aplicación personalizada.

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

Al ejecutar la actividad, las propiedades referenceObjects y extendedProperties se almacenan en los siguientes archivos que se implementan en la misma carpeta de ejecución del archivo SampleApp.exe:

activity.json

Almacena extendedProperties y las propiedades de la actividad personalizada.
linkedServices.json

Almacena una matriz de servicios vinculados definidos en la propiedad referenceObjects.
datasets.json

Almacena una matriz de conjuntos de datos definidos en la propiedad referenceObjects.

En el siguiente código de ejemplo se muestra cómo SampleApp.exe puede obtener acceso a la información necesaria de los archivos 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 los resultados de la ejecución

Puede iniciar una ejecución de canalización mediante el siguiente comando de PowerShell:

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

Durante la ejecución de la canalización, puede comprobar la salida de la ejecución mediante los comandos siguientes:

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"

Los archivos stdout y stderr de la aplicación personalizada se guardan en el contenedor adfjobs del servicio vinculado de Azure Storage definido al crear el servicio vinculado de Azure Batch con un GUID de la tarea. Puede obtener la ruta de acceso detallada de la salida de la ejecución de la actividad tal como se muestra en el siguiente fragmento de código:

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 desea usar el contenido de stdout.txt en actividades de bajada, puede obtener la ruta de acceso al archivo stdout.txt en la expresión "@activity('MyCustomActivity').output.outputs[0]".

Importante

Los archivos activity.json, linkedServices.json y datasets.json se almacenan en la carpeta de tiempo de ejecución de la tarea de Batch. En este ejemplo, los archivos activity.json, linkedServices.json y datasets.json se almacenan en la ruta de acceso https://adfv2storage.blob.core.windows.net/adfjobs/<GUID>/runtime/. Si es necesario, deberá limpiarlos por separado.
En cuanto a los servicios vinculados que usan un entorno de ejecución de integración autohospedado, la información confidencial (como claves o contraseñas) se cifra mediante el entorno de ejecución de integración autohospedado para garantizar que las credenciales permanecen en el entorno de red privada que haya definido el cliente. Si el código de la aplicación personalizada hace referencia de esta forma a algunos campos confidenciales, es posible que estos no estén presentes. Si es necesario, use SecureString en extendedProperties en lugar de usar la referencia al servicio vinculado.

Resultados del paso a otra actividad

Puede enviar los valores personalizados desde el código de una actividad personalizada al servicio. Puede hacerlo escribiéndolos en outputs.json desde la aplicación. El servicio copia el contenido de outputs.json y lo anexa a la salida de la actividad como el valor de la propiedad customOutput. (El límite de tamaño es de 2 MB). Si quiere consumir el contenido de outputs.json en las actividades posteriores, puede obtener el valor mediante la expresión @activity('<MyCustomActivity>').output.customOutput.

Recuperación de salidas de SecureString

Los valores de propiedades confidenciales designados como de tipo SecureString, tal y como se muestra en algunos de los ejemplos de este artículo, se enmascaran en la pestaña Supervisión de la interfaz de usuario. Sin embargo, en la ejecución de la canalización real, una propiedad SecureString se serializa como JSON dentro del archivo activity.json como texto sin formato. Por ejemplo:

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

Esta serialización no es verdaderamente segura y no está pensada para serlo. La intención es una sugerencia al servicio para enmascarar el valor en la pestaña Supervisión.

Para acceder a las propiedades de tipo SecureString desde una actividad personalizada, lea el archivo activity.json, que se coloca en la misma carpeta que el archivo .EXE, deserialice el archivo JSON y, a continuación, acceda a la propiedad JSON (extendedProperties => [propertyName] => valor).

Escalado automático de Azure Batch

También puede crear un grupo de Azure Batch con la característica autoescala . Por ejemplo, podría crear un grupo de Azure Batch con 0 máquinas virtuales dedicadas y una fórmula de escalado automático basada en el número de tareas pendientes.

La fórmula del ejemplo obtiene el comportamiento siguiente: Cuando el grupo se crea inicialmente, se inicia con una máquina virtual. La métrica $PendingTasks define el número de tareas que están en ejecución o activas (en cola). La fórmula busca el número promedio de tareas pendientes en los últimos 180 segundos y establece TargetDedicated en consecuencia. Garantiza que TargetDedicated nunca supera las 25 VM. Por tanto, a medida que se envían nuevas tareas, el grupo crece automáticamente y a medida que estás se completan, las VM se liberan una a una y el escalado automático las reduce. startingNumberOfVMs y maxNumberofVMs se pueden adaptar a sus necesidades.

Fórmula de escalado 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);

Para más información, consulte Escalado automático de los nodos de proceso en un grupo de Azure Batch .

Si el grupo usa el valor predeterminado de la propiedad autoScaleEvaluationInterval, el servicio Batch puede tardar de 15 a 30 minutos en preparar la máquina virtual antes de ejecutar la actividad personalizada. Si el grupo usa otro valor de autoScaleEvaluationInterval diferente, el servicio Batch podría tardar el valor de autoScaleEvaluationInterval más 10 minutos.

Vea los siguientes artículos, en los que se explica cómo transformar datos de otras maneras:

Uso de actividades personalizadas en una canalización de Azure Data Factory o Azure Synapse Analytics

Incorporación de actividades personalizadas a una canalización con la interfaz de usuario

Servicio vinculado de Azure Batch

Actividad personalizada

Permisos de la actividad personalizada

Ejecutar comandos

Pasar objetos y propiedades

Recuperar los resultados de la ejecución

Resultados del paso a otra actividad

Recuperación de salidas de SecureString

Escalado automático de Azure Batch

Contenido relacionado

Recursos adicionales