Używanie działań niestandardowych w potoku usługi Azure Data Factory lub Azure Synapse Analytics

Artykuł
12/09/2023

DOTYCZY: Azure Data Factory Azure Synapse Analytics

Napiwek

Wypróbuj usługę Data Factory w usłudze Microsoft Fabric — rozwiązanie analityczne typu all-in-one dla przedsiębiorstw. Usługa Microsoft Fabric obejmuje wszystko, od przenoszenia danych do nauki o danych, analizy w czasie rzeczywistym, analizy biznesowej i raportowania. Dowiedz się, jak bezpłatnie rozpocząć nową wersję próbną !

Istnieją dwa typy działań, których można użyć w potoku usługi Azure Data Factory lub Synapse.

Działania przenoszenia danych w celu przenoszenia danych między obsługiwanymi magazynami danych źródła i ujścia.
Działania przekształcania danych w celu przekształcania danych przy użyciu usług obliczeniowych, takich jak Azure HDInsight i Azure Batch.

Aby przenieść dane do/z magazynu danych, do którego usługa nie obsługuje, lub przekształcić/przetworzyć dane w sposób, który nie jest obsługiwany przez usługę, możesz utworzyć niestandardowe działanie z własną logiką przenoszenia lub przekształcania danych i używać działania w potoku. Działanie niestandardowe uruchamia dostosowaną logikę kodu w puli usługi Azure Batch maszyn wirtualnych.

Uwaga

Do interakcji z platformą Azure zalecamy używanie modułu Azure Az w programie PowerShell. Zobacz Instalowanie programu Azure PowerShell, aby rozpocząć. Aby dowiedzieć się, jak przeprowadzić migrację do modułu Az PowerShell, zobacz Migracja programu Azure PowerShell z modułu AzureRM do modułu Az.

Jeśli dopiero zaczynasz korzystać z usługi Azure Batch, zobacz następujące artykuły:

Podstawy usługi Azure Batch dotyczące przeglądu usługi Azure Batch.
Polecenie cmdlet New-AzBatchAccount w celu utworzenia konta usługi Azure Batch (lub) witryny Azure Portal w celu utworzenia konta usługi Azure Batch przy użyciu witryny Azure Portal. Aby uzyskać szczegółowe instrukcje dotyczące korzystania z polecenia cmdlet, zobacz artykuł Using PowerShell to manage Azure Batch Account (Zarządzanie kontem usługi Azure Batch przy użyciu programu PowerShell).
Polecenie cmdlet New-AzBatchPool w celu utworzenia puli usługi Azure Batch.

Ważne

Podczas tworzenia nowej puli usługi Azure Batch należy użyć polecenia "VirtualMachineConfiguration" i NIE "CloudServiceConfiguration". Aby uzyskać więcej informacji, zobacz Wskazówki dotyczące migracji puli usługi Azure Batch.

Dodawanie działań niestandardowych do potoku za pomocą interfejsu użytkownika

Aby użyć działania niestandardowego w potoku, wykonaj następujące kroki:

Wyszukaj ciąg Niestandardowy w okienku Działania potoku i przeciągnij działanie niestandardowe na kanwę potoku.
Wybierz nowe działanie Niestandardowe na kanwie, jeśli nie zostało jeszcze wybrane.
Wybierz kartę Azure Batch , aby wybrać lub utworzyć nową połączoną usługę Azure Batch, która wykona działanie niestandardowe.
Wybierz kartę Ustawienia i określ polecenie do wykonania w usłudze Azure Batch i opcjonalne szczegóły zaawansowane.

Połączona usługa Azure Batch

Poniższy kod JSON definiuje przykładową połączoną usługę Azure Batch. Aby uzyskać szczegółowe informacje, zobacz Obsługiwane środowiska obliczeniowe

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

Aby dowiedzieć się więcej o połączonej usłudze Azure Batch, zobacz artykuł Compute linked services (Połączone usługi obliczeniowe).

Działanie niestandardowe

Poniższy fragment kodu JSON definiuje potok z prostym działaniem niestandardowym. Definicja działania zawiera odwołanie do połączonej usługi 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"
        }
      }
    }]
  }
}

W tym przykładzie plik helloworld.exe to aplikacja niestandardowa przechowywana w folderze customactv2/helloworld konta usługi Azure Storage używanego w usłudze resourceLinkedService. Działanie niestandardowe przesyła tę aplikację niestandardową do wykonania w usłudze Azure Batch. Możesz zastąpić polecenie dowolną preferowaną aplikacją, która może być wykonywana w docelowym systemie operacyjnym węzłów puli usługi Azure Batch.

W poniższej tabeli opisano nazwy i opisy właściwości specyficznych dla tego działania.

Właściwości	Opis	Wymagania
name	Nazwa działania w potoku	Tak
opis	Tekst opisujący działanie.	Nie.
type	W przypadku działania niestandardowego typ działania to Niestandardowy.	Tak
linkedServiceName	Połączona usługa z usługą Azure Batch. Aby dowiedzieć się więcej o tej połączonej usłudze, zobacz artykuł Dotyczący połączonych usług obliczeniowych.	Tak
polecenie	Polecenie aplikacji niestandardowej do wykonania. Jeśli aplikacja jest już dostępna w węźle puli usługi Azure Batch, można pominąć parametr resourceLinkedService i folderPath. Można na przykład określić polecenie na `cmd /c dir`wartość , które jest natywnie obsługiwane przez węzeł Puli usługi Windows Batch.	Tak
resourceLinkedService	Połączona usługa Azure Storage z kontem magazynu, na którym jest przechowywana aplikacja niestandardowa	Nr*
folderPath	Ścieżka do folderu aplikacji niestandardowej i wszystkich jej zależności Jeśli masz zależności przechowywane w podfolderach — czyli w strukturze folderów hierarchicznych w folderPath — struktura folderów jest obecnie spłaszczana, gdy pliki są kopiowane do usługi Azure Batch. Oznacza to, że wszystkie pliki są kopiowane do jednego folderu bez podfolderów. Aby obejść to zachowanie, rozważ skompresowanie plików, skopiowanie skompresowanego pliku, a następnie rozpakowanie go przy użyciu kodu niestandardowego w żądanej lokalizacji.	Nr*
referenceObjects	Tablica istniejących połączonych usług i zestawów danych. Przywoływane połączone usługi i zestawy danych są przekazywane do aplikacji niestandardowej w formacie JSON, dzięki czemu kod niestandardowy może odwoływać się do zasobów usługi	Nie.
Extendedproperties	Właściwości zdefiniowane przez użytkownika, które można przekazać do aplikacji niestandardowej w formacie JSON, aby kod niestandardowy mógł odwoływać się do dodatkowych właściwości	Nie.
retentionTimeInDays	Czas przechowywania plików przesłanych do działania niestandardowego. Wartość domyślna to 30 dni.	Nie.

* Właściwości resourceLinkedService i folderPath muszą zostać określone albo oba te właściwości zostaną pominięte.

Uwaga

Jeśli przekazujesz połączone usługi jako referenceObjects w działaniu niestandardowym, dobrym rozwiązaniem jest przekazanie połączonej usługi Azure Key Vault włączonej (ponieważ nie zawiera żadnych bezpiecznych ciągów) i pobranie poświadczeń przy użyciu nazwy wpisu tajnego bezpośrednio z usługi Key Vault z kodu. Przykład można znaleźć tutaj , który odwołuje się do połączonej usługi z włączoną usługą AKV, pobiera poświadczenia z usługi Key Vault, a następnie uzyskuje dostęp do magazynu w kodzie.

Uwaga

Obecnie tylko usługa Azure Blob Storage jest obsługiwana dla zasobuLinkedService w działaniu niestandardowym i jest to jedyna połączona usługa, która jest tworzona domyślnie i nie ma opcji wyboru innych łączników, takich jak ADLS Gen2.

Uprawnienia do działań niestandardowych

Niestandardowe działanie ustawia automatyczne konto użytkownika usługi Azure Batch na dostęp bez uprawnień administratora z zakresem zadań (domyślna specyfikacja automatycznego użytkownika). Nie można zmienić poziomu uprawnień konta użytkownika automatycznego. Aby uzyskać więcej informacji, zobacz Uruchamianie zadań w ramach kont użytkowników w usłudze Batch | Konta użytkowników automatycznych.

Wykonywanie poleceń

Polecenie można wykonać bezpośrednio przy użyciu działania niestandardowego. Poniższy przykład uruchamia polecenie "echo hello world" w docelowych węzłach puli usługi Azure Batch i wyświetla dane wyjściowe do 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"
      }
    }]
  }
}

Przekazywanie obiektów i właściwości

W tym przykładzie pokazano, jak za pomocą obiektów referenceObjects i extendedProperties przekazywać obiekty i właściwości zdefiniowane przez użytkownika z usługi do aplikacji niestandardowej.

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

Po wykonaniu działania obiekty referenceObjects i extendedProperties są przechowywane w następujących plikach wdrożonych w tym samym folderze wykonywania pliku SampleApp.exe:

activity.json

Przechowuje właściwości rozszerzone i właściwości działania niestandardowego.
linkedServices.json

Przechowuje tablicę połączonych usług zdefiniowanych we właściwości referenceObjects.
datasets.json

Przechowuje tablicę zestawów danych zdefiniowanych we właściwości referenceObjects.

Poniższy przykładowy kod pokazuje, jak plik SampleApp.exe może uzyskać dostęp do wymaganych informacji z plików 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);
        }
    }
}

Pobieranie danych wyjściowych wykonywania

Uruchomienie potoku można uruchomić przy użyciu następującego polecenia programu PowerShell:

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

Po uruchomieniu potoku możesz sprawdzić dane wyjściowe wykonywania przy użyciu następujących poleceń:

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 i stderr aplikacji niestandardowej są zapisywane w kontenerze adfjobs w połączonej usłudze Azure Storage zdefiniowanej podczas tworzenia połączonej usługi Azure Batch z identyfikatorem GUID zadania. Szczegółową ścieżkę można uzyskać z danych wyjściowych uruchomienia działania, jak pokazano w poniższym fragmencie kodu:

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"

Jeśli chcesz użyć zawartości pliku stdout.txt w działaniach podrzędnych, możesz uzyskać ścieżkę do pliku stdout.txt w wyrażeniu "@activity('MyCustomActivity').outputs[0]".

Ważne

Plik activity.json, linkedServices.json i datasets.json są przechowywane w folderze uruchomieniowym zadania usługi Batch. W tym przykładzie plik activity.json, linkedServices.json i datasets.json są przechowywane w https://adfv2storage.blob.core.windows.net/adfjobs/<GUID>/runtime/ ścieżce. W razie potrzeby należy je wyczyścić oddzielnie.
W przypadku połączonych usług korzystających z własnego środowiska Integration Runtime poufne informacje, takie jak klucze lub hasła, są szyfrowane przez własne środowisko Integration Runtime w celu zapewnienia, że poświadczenia pozostaną w środowisku sieci prywatnej zdefiniowanej przez klienta. Niektórych poufnych pól może brakować, jeśli w ten sposób odwołuje się do kodu aplikacji niestandardowej. W razie potrzeby użyj funkcji SecureString w właściwościach extendedProperties zamiast używania dokumentacji połączonej usługi.

Przekazywanie danych wyjściowych do innego działania

Możesz wysyłać wartości niestandardowe z kodu w działaniu niestandardowym z powrotem do usługi. Możesz to zrobić, zapisując je w outputs.json aplikacji. Usługa kopiuje zawartość outputs.json elementu i dołącza ją do danych wyjściowych działania jako wartość customOutput właściwości . (Limit rozmiaru to 2 MB). Jeśli chcesz korzystać z zawartości w działaniach podrzędnych outputs.json , możesz uzyskać wartość przy użyciu wyrażenia @activity('<MyCustomActivity>').output.customOutput.

Pobieranie danych wyjściowych funkcji SecureString

Poufne wartości właściwości wyznaczone jako typ SecureString, jak pokazano w niektórych przykładach w tym artykule, są maskowane na karcie Monitorowanie w interfejsie użytkownika. Jednak w rzeczywistym wykonaniu potoku właściwość SecureString jest serializowana jako plik JSON w activity.json pliku jako zwykły tekst. Na przykład:

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

Ta serializacja nie jest naprawdę bezpieczna i nie ma być bezpieczna. Intencją jest wskazówka dla usługi maskowania wartości na karcie Monitorowanie.

Aby uzyskać dostęp do właściwości typu SecureString z działania niestandardowego, przeczytaj activity.json plik, który znajduje się w tym samym folderze co plik EXE, deserializuj kod JSON, a następnie uzyskaj dostęp do właściwości JSON (extendedProperties => [propertyName] => wartość).

Automatyczne skalowanie usługi Azure Batch

Możesz również utworzyć pulę usługi Azure Batch z funkcją automatycznego skalowania . Można na przykład utworzyć pulę wsadową platformy Azure z 0 dedykowanymi maszynami wirtualnymi i formułą autoskalowania na podstawie liczby oczekujących zadań.

Przykładowa formuła w tym miejscu osiąga następujące zachowanie: po początkowym utworzeniu puli rozpoczyna się od 1 maszyny wirtualnej. $PendingTasks metryka definiuje liczbę zadań w stanie uruchamiania i aktywnego (w kolejce). Formuła znajduje średnią liczbę oczekujących zadań w ciągu ostatnich 180 sekund i ustawia wartość TargetDedicated odpowiednio. Gwarantuje to, że targetDedicated nigdy nie wykracza poza 25 maszyn wirtualnych. W miarę przesyłania nowych zadań pula automatycznie rośnie i w miarę wykonywania zadań maszyny wirtualne stają się wolne od jednego do jednego, a skalowanie automatyczne zmniejsza te maszyny wirtualne. startNumberOfVMs i maxNumberofVMs można dostosować do Twoich potrzeb.

Formuła autoskalu:

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

Aby uzyskać szczegółowe informacje, zobacz Automatyczne skalowanie węzłów obliczeniowych w puli usługi Azure Batch.

Jeśli pula używa domyślnego autoScaleEvaluationInterval, usługa Batch może potrwać 15–30 minut, aby przygotować maszynę wirtualną przed uruchomieniem działania niestandardowego. Jeśli pula używa innego autoScaleEvaluationInterval, usługa Batch może potrwać autoScaleEvalInterval + 10 minut.

Zapoznaj się z następującymi artykułami, które wyjaśniają sposób przekształcania danych na inne sposoby:

Używanie działań niestandardowych w potoku usługi Azure Data Factory lub Azure Synapse Analytics

Dodawanie działań niestandardowych do potoku za pomocą interfejsu użytkownika

Połączona usługa Azure Batch

Działanie niestandardowe

Uprawnienia do działań niestandardowych

Wykonywanie poleceń

Przekazywanie obiektów i właściwości

Pobieranie danych wyjściowych wykonywania

Przekazywanie danych wyjściowych do innego działania

Pobieranie danych wyjściowych funkcji SecureString

Automatyczne skalowanie usługi Azure Batch

Powiązana zawartość

Dodatkowe zasoby