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

PLATÍ PRO: Azure Data Factory Azure Synapse Analytics

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 .

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

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:

1. Vyhledejte vlastní v podokně Aktivity kanálu a přetáhněte vlastní aktivitu na plátno kanálu.

2. Pokud ještě není vybraná, vyberte na plátně novou vlastní aktivitu.

3. Výběrem karty Azure Batch vyberte nebo vytvořte novou propojenou službu Azure Batch, která spustí vlastní aktivitu.

   

4. 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.

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

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 získat 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 využívat obsah stdout.txt v podřízených aktivitách, můžete získat cestu k souboru stdout.txt ve výrazu "@activity('MyCustomActivity').outputs[0]".

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 váš .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.

Související obsah

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

• Aktivita U-SQL
• Aktivita Hivu
• Aktivita Pig
• Aktivita MapReduce
• Aktivita streamování Hadoop
• Aktivita Sparku
• Aktivita uložená procedura