Použití vlastních aktivit v kanálu Azure Data Factory nebo Azure Synapse Analytics

Článek
12/09/2023

PLATÍ PRO: Azure Data Factory Azure Synapse Analytics

Tip

Vyzkoušejte si službu Data Factory v Microsoft Fabric, řešení pro analýzy typu all-in-one pro podniky. Microsoft Fabric zahrnuje všechno od přesunu dat až po datové vědy, analýzy v reálném čase, business intelligence a vytváření sestav. Přečtěte si, jak začít používat novou zkušební verzi zdarma.

Existují dva typy aktivit, které můžete použít v kanálu Azure Data Factory nebo Synapse.

Aktivity přesunu dat pro přesun dat mezi podporovaným zdrojovým úložištěm a úložišti dat jímky
Aktivity transformace dat pro transformaci dat pomocí výpočetních služeb, jako jsou Azure HDInsight a Azure Batch.

Pokud chcete přesunout data do a z úložiště dat, které služba nepodporuje, nebo transformovat a zpracovávat data způsobem, který služba nepodporuje, můžete vytvořit vlastní aktivitu s vlastní logikou přesunu nebo transformace dat a použít aktivitu v kanálu. Vlastní aktivita spouští vlastní logiku kódu ve fondu virtuálních počítačů Azure Batch .

Poznámka:

Při práci s Azure doporučujeme používat modul Azure Az PowerShellu. Začněte tím, že si projdete téma Instalace Azure PowerShellu. Informace o tom, jak migrovat na modul Az PowerShell, najdete v tématu Migrace Azure PowerShellu z AzureRM na Az.

Pokud s službou Azure Batch začínáte, projděte si následující články:

Základy služby Azure Batch pro přehled služby Azure Batch
Rutina New-AzBatchAccount pro vytvoření účtu Azure Batch (nebo webu Azure Portal) pro vytvoření účtu Azure Batch pomocí webu Azure Portal. Podrobné pokyny k používání rutiny najdete v článku o správě účtu Azure Batch pomocí PowerShellu.
Rutina New-AzBatchPool pro vytvoření fondu Azure Batch

Důležité

Při vytváření nového fondu Azure Batch se musí použít VirtualMachineConfiguration a NE CloudServiceConfiguration. Další podrobnosti najdete v pokynech k migraci fondu Azure Batch.

Přidání vlastních aktivit do kanálu pomocí uživatelského rozhraní

Pokud chcete v kanálu použít vlastní aktivitu, proveďte následující kroky:

Vyhledejte vlastní v podokně Aktivity kanálu a přetáhněte vlastní aktivitu na plátno kanálu.
Pokud ještě není vybraná, vyberte na plátně novou vlastní aktivitu.
Výběrem karty Azure Batch vyberte nebo vytvořte novou propojenou službu Azure Batch, která spustí vlastní aktivitu.
Vyberte kartu Nastavení a zadejte příkaz, který se má spustit ve službě Azure Batch, a volitelné pokročilé podrobnosti.

Propojená služba Azure Batch

Následující JSON definuje ukázkovou propojenou službu Azure Batch. Podrobnosti najdete v tématu Podporovaná výpočetní prostředí.

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

Další informace o propojené službě Azure Batch najdete v článku o propojených službách Compute.

Vlastní aktivita

Následující fragment kódu JSON definuje kanál s jednoduchou vlastní aktivitou. Definice aktivity obsahuje odkaz na propojenou službu 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"
        }
      }
    }]
  }
}

V této ukázce je helloworld.exe vlastní aplikace uložená ve složce customactv2/helloworld účtu Azure Storage použitého v resourceLinkedService. Vlastní aktivita odešle tuto vlastní aplikaci ke spuštění ve službě Azure Batch. Příkaz můžete nahradit jakoukoli upřednostňovanou aplikací, která se dá spustit v cílovém operačním systému uzlů fondu Azure Batch.

Následující tabulka popisuje názvy a popisy vlastností, které jsou specifické pro tuto aktivitu.

Vlastnost	Popis	Povinní účastníci
name	Název aktivity v kanálu	Ano
description	Text popisující, co aktivita dělá.	No
type	Pro vlastní aktivitu je typ aktivity Vlastní.	Ano
linkedServiceName	Propojená služba se službou Azure Batch Další informace o této propojené službě najdete v článku o propojených službách Compute.	Ano
příkaz	Příkaz vlastní aplikace, která se má spustit. Pokud je aplikace již dostupná na uzlu fondu služby Azure Batch, je možné přeskočit resourceLinkedService a folderPath. Můžete například zadat příkaz, který má být `cmd /c dir`nativně podporován uzlem fondu služby Windows Batch.	Ano
resourceLinkedService	Propojená služba Azure Storage s účtem úložiště, ve kterém je uložená vlastní aplikace	Ne*
folderPath	Cesta ke složce vlastní aplikace a všechny její závislosti Pokud máte závislosti uložené v podsložkách – to znamená v hierarchické struktuře složek v cestě folderPath – struktura složek se v současné době zploštěla při kopírování souborů do služby Azure Batch. To znamená, že všechny soubory se zkopírují do jedné složky bez podsložek. Pokud chcete toto chování obejít, zvažte komprimaci souborů, zkopírování komprimovaného souboru a jeho rozbalení vlastním kódem v požadovaném umístění.	Ne*
referenceObjects	Pole existujících propojených služeb a datových sad. Odkazované propojené služby a datové sady se předávají vlastní aplikaci ve formátu JSON, aby váš vlastní kód mohl odkazovat na prostředky služby.	No
extendedProperties	Uživatelem definované vlastnosti, které je možné předat vlastní aplikaci ve formátu JSON, aby váš vlastní kód mohl odkazovat na další vlastnosti	No
retentionTimeInDays	Doba uchovávání souborů odeslaných pro vlastní aktivitu. Výchozí hodnota je 30 dní.	No

* Vlastnosti resourceLinkedService a folderPath musí být buď zadány, nebo oba musí být vynechány.

Poznámka:

Pokud předáváte propojené služby jako referenceObjects ve vlastní aktivitě, je vhodné předat propojenou službu s povolenou službou Azure Key Vault (protože neobsahuje žádné zabezpečené řetězce) a načíst přihlašovací údaje pomocí názvu tajného kódu přímo ze služby Key Vault. Tady najdete příklad, který odkazuje na propojenou službu s podporou AKV, načte přihlašovací údaje ze služby Key Vault a pak přistupuje k úložišti v kódu.

Poznámka:

V současné době se ve vlastní aktivitě podporuje pouze azure Blob Storage pro resourceLinkedService a jedná se o jedinou propojenou službu, která se ve výchozím nastavení vytvoří a není možné zvolit jiné konektory, jako je ADLS Gen2.

Vlastní oprávnění aktivit

Vlastní aktivita nastaví účet automatického uživatele Služby Azure Batch na přístup bez oprávnění správce s oborem úkolu (výchozí specifikace automatického uživatele). Úroveň oprávnění automatického uživatelského účtu nemůžete změnit. Další informace najdete v tématu Spouštění úloh v uživatelských účtech ve službě Batch | Automatické uživatelské účty.

Spouštění příkazů

Příkaz můžete spustit přímo pomocí vlastní aktivity. Následující příklad spustí příkaz "echo hello world" na cílových uzlech fondu Azure Batch a vytiskne výstup do stdoutu.

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

Předávání objektů a vlastností

Tato ukázka ukazuje, jak můžete použít referenceObjects a extendedProperties k předávání objektů a uživatelem definovaných vlastností ze služby do vlastní aplikace.

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

Při spuštění aktivity jsou referenceObjects a extendedProperties uloženy v následujících souborech, které jsou nasazeny do stejné spouštěcí složky SampleApp.exe:

activity.json

Ukládá extendedProperties a vlastnosti vlastní aktivity.
linkedServices.json

Uloží pole propojených služeb definovaných ve vlastnosti referenceObjects.
datasets.json

Uloží pole datových sad definovaných ve vlastnosti referenceObjects.

Následující ukázkový kód ukazuje, jak SampleApp.exe má přístup k požadovaným informacím ze souborů 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);
        }
    }
}

Načtení výstupů spuštění

Spuštění kanálu můžete spustit pomocí následujícího příkazu PowerShellu:

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

Když je kanál spuštěný, můžete zkontrolovat výstup spuštění pomocí následujících příkazů:

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"

Stdout a stderr vaší vlastní aplikace se ukládají do kontejneru adfjobs v propojené službě Azure Storage, kterou jste definovali při vytváření propojené služby Azure Batch s identifikátorem GUID úlohy. Podrobnou cestu můžete získat z výstupu spuštění aktivity, jak je znázorněno v následujícím fragmentu kódu:

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"

Pokud chcete použít obsah souboru stdout.txt v podřízených aktivitách, můžete získat cestu k souboru stdout.txt ve výrazu "@activity('MyCustomActivity').output.outputs[0]".

Důležité

Soubor activity.json, linkedServices.json a datasets.json jsou uložené ve složce runtime úlohy Batch. V tomto příkladu jsou v cestě uloženy https://adfv2storage.blob.core.windows.net/adfjobs/<GUID>/runtime/ soubory activity.json, linkedServices.json a datasets.json. V případě potřeby je potřebujete vyčistit samostatně.
U propojených služeb, které používají místní prostředí Integration Runtime, jsou citlivé informace, jako jsou klíče nebo hesla, šifrované modulem Integration Runtime v místním prostředí, aby se zajistilo, že přihlašovací údaje zůstanou v prostředí privátní sítě definované zákazníkem. Při odkazování vlastním kódem aplikace tímto způsobem může chybět některá citlivá pole. V případě potřeby používejte řetězec SecureString v rozšířenýchpropertiesch místo odkazu na propojenou službu.

Předání výstupů jiné aktivitě

Vlastní hodnoty z kódu ve vlastní aktivitě můžete odeslat zpět do služby. Můžete to udělat tak, že je napíšete z outputs.json vaší aplikace. Služba zkopíruje obsah a outputs.json připojí ho do výstupu customOutput aktivity jako hodnotu vlastnosti. (Limit velikosti je 2 MB.) Pokud chcete využívat obsah podřízených outputs.json aktivit, můžete hodnotu získat pomocí výrazu @activity('<MyCustomActivity>').output.customOutput.

Načtení výstupů securestringu

Hodnoty citlivých vlastností označené jako typ SecureString, jak je znázorněno v některých příkladech v tomto článku, jsou maskovány na kartě Monitorování v uživatelském rozhraní. Ve skutečném spuštění kanálu je však vlastnost SecureString serializována jako JSON v souboru activity.json jako prostý text. Příklad:

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

Tato serializace není skutečně zabezpečená a není určena k zabezpečení. Záměrem je nápověda pro službu, která má maskovat hodnotu na kartě Monitorování.

Pokud chcete získat přístup k vlastnostem typu SecureString z vlastní aktivity, přečtěte si activity.json soubor, který je umístěný ve stejné složce jako .EXE, deserializovat JSON a pak získat přístup k vlastnosti JSON (extendedProperties => [propertyName] => value).

Automatické škálování služby Azure Batch

Můžete také vytvořit fond Azure Batch s funkcí automatického škálování . Můžete například vytvořit fond Azure Batch s 0 vyhrazenými virtuálními počítači a vzorcem automatického škálování na základě počtu čekajících úkolů.

Ukázkový vzorec zde dosahuje následujícího chování: Při počátečním vytvoření fondu začíná 1 virtuálním počítačem. $PendingTasks metrika definuje počet úloh ve stavu spuštěno + aktivní (ve frontě). Vzorec najde průměrný počet čekajících úkolů za posledních 180 sekund a nastaví targetDedicated odpovídajícím způsobem. Zajišťuje, aby TargetDedicated nikdy nepřesahoval 25 virtuálních počítačů. Když se tedy odesílají nové úkoly, fond se automaticky zvětší a po dokončení úkolů se virtuální počítače uvolní o jeden po druhém a automatické škálování tyto virtuální počítače zmenší. hodnoty startNumberOfVMs a maxNumberofVMs je možné upravit podle vašich potřeb.

Vzorec automatického škálování:

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

Podrobnosti najdete v tématu Automatické škálování výpočetních uzlů ve fondu Azure Batch.

Pokud fond používá výchozí automatické škálováníIntervaluationInterval, může příprava virtuálního počítače před spuštěním vlastní aktivity trvat 15 až 30 minut. Pokud fond používá jiný autoScaleEvaluationInterval, může služba Batch trvat automatické škálováníEvaluationInterval + 10 minut.

Podívejte se na následující články, které vysvětlují, jak transformovat data jinými způsoby:

Použití vlastních aktivit v kanálu Azure Data Factory nebo Azure Synapse Analytics

Přidání vlastních aktivit do kanálu pomocí uživatelského rozhraní

Propojená služba Azure Batch

Vlastní aktivita

Vlastní oprávnění aktivit

Spouštění příkazů

Předávání objektů a vlastností

Načtení výstupů spuštění

Předání výstupů jiné aktivitě

Načtení výstupů securestringu

Automatické škálování služby Azure Batch

Související obsah

Další materiály