Szybki start: Tworzenie fabryki danych i potoku przy użyciu zestawu SDK .NET

Artykuł
03/09/2023
Czas czytania: 12 min

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym przewodniku Szybki start opisano sposób tworzenia Azure Data Factory przy użyciu zestawu SDK platformy .NET. Potok tworzony w tej fabryce danych kopiuje dane z jednego folderu do innego folderu w usłudze Azure Blob Storage. Aby zapoznać się z samouczkiem dotyczącym przekształcania danych przy użyciu Azure Data Factory, zobacz Samouczek: przekształcanie danych przy użyciu platformy Spark.

Uwaga

Ten artykuł nie zawiera szczegółowego wprowadzenia do usługi Data Factory. Aby zapoznać się z wprowadzeniem do usługi Azure Data Factory, zobacz Wprowadzenie do usługi Azure Data Factory.

Wymagania wstępne

Subskrypcja platformy Azure

Jeśli nie masz subskrypcji platformy Azure, przed rozpoczęciem utwórz bezpłatne konto.

Role platformy Azure

Aby utworzyć wystąpienia usługi Data Factory, konto użytkownika używane do logowania się na platformie Azure musi być członkiem roli współautora lub właściciela albo administratorem subskrypcji platformy Azure. Aby wyświetlić uprawnienia, które masz w subskrypcji, przejdź do Azure Portal, wybierz swoją nazwę użytkownika w prawym górnym rogu, wybierz ikonę "...", aby uzyskać więcej opcji, a następnie wybierz pozycję Moje uprawnienia. Jeśli masz dostęp do wielu subskrypcji, wybierz odpowiednią subskrypcję.

Aby utworzyć zasoby podrzędne — w tym zestawy danych, połączone usługi, potoki, wyzwalacze i środowiska Integration Runtime — oraz nimi zarządzać w usłudze Data Factory, należy spełnić następujące wymagania:

Aby utworzyć zasoby podrzędne i zarządzać nimi w Azure Portal, musisz należeć do roli Współautor usługi Data Factory na poziomie grupy zasobów lub wyższym.
W przypadku tworzenia zasobów podrzędnych i zarządzania nimi za pomocą programu PowerShell lub zestawu SDK rola współautora na poziomie grupy zasobów lub wyższej jest wystarczająca.

Aby uzyskać przykładowe instrukcje dotyczące dodawania użytkownika do roli, zobacz artykuł Add roles (Dodawanie ról).

Aby uzyskać więcej informacji, zobacz następujące artykuły:

Konto usługi Azure Storage

W tym przewodniku Szybki start użyjesz konta usługi Azure Storage ogólnego przeznaczenia (w szczególności usługi Blob Storage) jako magazynów danych źródłowych i docelowych . Jeśli nie masz konta usługi Azure Storage ogólnego przeznaczenia, zobacz Tworzenie konta magazynu , aby je utworzyć.

Pobieranie nazwy konta magazynu

Do tego przewodnika Szybki start potrzebna jest nazwa konta usługi Azure Storage. Poniższa procedura zawiera kroki umożliwiające uzyskanie nazwy konta magazynu:

W przeglądarce internetowej przejdź do Azure Portal i zaloguj się przy użyciu nazwy użytkownika i hasła platformy Azure.
W menu Azure Portal wybierz pozycję Wszystkie usługi, a następnie wybierz pozycjęKonta magazynu magazynu>. Możesz również wyszukać i wybrać pozycję Konta magazynu na dowolnej stronie.
Na stronie Konta magazynu przefiltruj konto magazynu (w razie potrzeby), a następnie wybierz swoje konto magazynu.

Możesz również wyszukać i wybrać pozycję Konta magazynu na dowolnej stronie.

Tworzenie kontenera obiektów blob

W tej sekcji utworzysz kontener obiektów blob o nazwie adftutorial w usłudze Azure Blob Storage.

Na stronie konta magazynu wybierz pozycję Kontenery przeglądów>.
Na pasku narzędzikontenerównazwy>konta< wybierzpozycję Kontener. -
W oknie dialogowym Nowy kontener wprowadź jako nazwę adftutorial, a następnie wybierz przycisk OK. Strona <Kontenery nazw> - kont została zaktualizowana w celu uwzględnienia elementu adftutorial na liście kontenerów.

Dodawanie folderu wejściowego i pliku dla kontenera obiektów blob

W tej sekcji utworzysz folder o nazwie input w utworzonym kontenerze, a następnie przekażesz przykładowy plik do folderu wejściowego. Przed rozpoczęciem otwórz edytor tekstów, taki jak Notatnik, i utwórz plik o nazwie emp.txt z następującą zawartością:

John, Doe
Jane, Doe

Zapisz plik w folderze C:\ADFv2QuickStartPSH . (Jeśli folder jeszcze nie istnieje, utwórz go). Następnie wróć do Azure Portal i wykonaj następujące kroki:

Na stronieKontenery nazwy - >konta, na której została wyłączona, wybierz pozycję adftutorial ze zaktualizowanej listy kontenerów<.
1. Jeśli okno zostało zamknięte lub poszło na inną stronę, zaloguj się ponownie do Azure Portal.
2. W menu Azure Portal wybierz pozycję Wszystkie usługi, a następnie wybierz pozycjęKonta magazynu magazynu>. Możesz również wyszukać i wybrać pozycję Konta magazynu na dowolnej stronie.
3. Wybierz konto magazynu, a następnie wybierz pozycję Kontenery>adftutorial.
Na pasku narzędzi strony kontenera adftutorial wybierz pozycję Przekaż.
Na stronie Przekazywanie obiektu blob wybierz pole Pliki , a następnie przejdź do i wybierz plik emp.txt .
Rozwiń nagłówek Zaawansowane . Strona jest teraz wyświetlana w następujący sposób:
W polu Przekaż do folderu wprowadź dane wejściowe.
Wybierz przycisk Przekaż. Na liście powinien pojawić się plik emp.txt i stan przekazywania.
Wybierz ikonę Zamknij(x), aby zamknąć stronę Przekaż obiekt blob .

Pozostaw otwartą stronę kontenera adftutorial . Będzie ona używana do weryfikowania danych wyjściowych na końcu tego samouczka Szybki start.

Visual Studio

Przewodnik w tym artykule używa programu Visual Studio 2019. Procedury dotyczące Visual Studio 2013, 2015 lub 2017 różnią się nieznacznie.

Utworzenie aplikacji w usłudze Azure Active Directory

W sekcjach w temacie Instrukcje: tworzenie aplikacji i jednostki usługi Azure AD za pomocą portalu, które mogą uzyskiwać dostęp do zasobów, postępuj zgodnie z instrukcjami, aby wykonać następujące zadania:

W obszarze Tworzenie aplikacji usługi Azure Active Directory utwórz aplikację reprezentującą aplikację platformy .NET, którą tworzysz w tym samouczku. W przypadku adresu URL logowania możesz podać fikcyjny adres URL, jak pokazano w artykule (https://contoso.org/exampleapp).
W obszarze Pobierz wartości logowania pobierz identyfikator aplikacji i identyfikator dzierżawy, a następnie zanotuj te wartości, które będą używane w dalszej części tego samouczka.
W obszarze Certyfikaty i wpisy tajne pobierz klucz uwierzytelniania i zanotuj tę wartość, która zostanie użyta w dalszej części tego samouczka.
W obszarze Przypisywanie aplikacji do roli przypisz aplikację do roli Współautor na poziomie subskrypcji, aby aplikacja mogła tworzyć fabryki danych w subskrypcji.

Tworzenie projektu programu Visual Studio

Następnie utwórz aplikację konsolową platformy .NET w języku C# w programie Visual Studio:

Uruchom program Visual Studio.
W oknie Start wybierz pozycję Utwórz nową aplikację konsolową projektu>(.NET Framework). Wymagana jest platforma .NET w wersji 4.5.2 lub nowszej.
W polu Nazwa projektu wprowadź wartość ADFv2QuickStart.
Wybierz polecenie Create (Utwórz), aby utworzyć projekt.

Instalowanie pakietów NuGet

Wybierz pozycję Narzędzia>Menedżer pakietów NuGet>Konsola menedżera pakietów.
W okienku Konsola menedżera pakietów uruchom następujące polecenia, aby zainstalować pakiety. Aby uzyskać więcej informacji, zobacz pakiet NuGet Microsoft.Azure.Management.DataFactory.
```
Install-Package Microsoft.Azure.Management.DataFactory
Install-Package Microsoft.Azure.Management.ResourceManager -IncludePrerelease
Install-Package Microsoft.Identity.Client
```

Tworzenie klienta fabryki danych

Otwórz plik Program.cs, a następnie dołącz poniższe instrukcje, aby dodać odwołania do przestrzeni nazw.

using System;
using System.Collections.Generic;
using System.Linq;
using Microsoft.Rest;
using Microsoft.Rest.Serialization;
using Microsoft.Azure.Management.ResourceManager;
using Microsoft.Azure.Management.DataFactory;
using Microsoft.Azure.Management.DataFactory.Models;
using Microsoft.Identity.Client;

Dodaj do metody Main następujący kod, który określa zmienne. Zastąp symbole zastępcze własnymi wartościami. Aby uzyskać listę regionów platformy Azure, w których obecnie jest dostępna usługa Data Factory, wybierz dane regiony na poniższej stronie, a następnie rozwiń węzeł Analiza, aby zlokalizować pozycję Data Factory: Produkty dostępne według regionu. Magazyny danych (Azure Storage, Azure SQL Database i inne) i obliczenia (HDInsight i inne) używane przez fabrykę danych mogą znajdować się w innych regionach.

// Set variables
string tenantID = "<your tenant ID>";
string applicationId = "<your application ID>";
string authenticationKey = "<your authentication key for the application>";
string subscriptionId = "<your subscription ID where the data factory resides>";
string resourceGroup = "<your resource group where the data factory resides>";
string region = "<the location of your resource group>";
string dataFactoryName = 
    "<specify the name of data factory to create. It must be globally unique.>";
string storageAccount = "<your storage account name to copy data>";
string storageKey = "<your storage account key>";
// specify the container and input folder from which all files 
// need to be copied to the output folder. 
string inputBlobPath =
    "<path to existing blob(s) to copy data from, e.g. containername/inputdir>";
//specify the contains and output folder where the files are copied
string outputBlobPath =
    "<the blob path to copy data to, e.g. containername/outputdir>";

// name of the Azure Storage linked service, blob dataset, and the pipeline
string storageLinkedServiceName = "AzureStorageLinkedService";
string blobDatasetName = "BlobDataset";
string pipelineName = "Adfv2QuickStartPipeline";

Uwaga

W przypadku chmur suwerennych należy użyć odpowiednich punktów końcowych specyficznych dla chmury dla usług ActiveDirectoryAuthority i ResourceManagerUrl (BaseUri). Na przykład w usłudze Azure Gov w STANACH Zjednoczonych należy użyć https://management.usgovcloudapi.net urzędu https://login.microsoftonline.us zamiast https://login.microsoftonline.com https://management.azure.com/, a nie , a następnie utworzyć klienta zarządzania fabryką danych. Za pomocą programu PowerShell można łatwo uzyskać adresy URL punktów końcowych dla różnych chmur, wykonując polecenie "Get-AzEnvironment | Format-List", który zwróci listę punktów końcowych dla każdego środowiska chmury.

Dodaj następujący kod do metody Main , która tworzy wystąpienie klasy DataFactoryManagementClient . Ten obiekt jest używany do tworzenia fabryki danych, połączonej usługi, zestawów danych i potoku. Umożliwia on również monitorowanie szczegółów uruchomienia potoku.

// Authenticate and create a data factory management client
IConfidentialClientApplication app = ConfidentialClientApplicationBuilder.Create(applicationId)
 .WithAuthority("https://login.microsoftonline.com/" + tenantID)
 .WithClientSecret(authenticationKey)
 .WithLegacyCacheCompatibility(false)
 .WithCacheOptions(CacheOptions.EnableSharedCacheOptions)
 .Build();

AuthenticationResult result = await app.AcquireTokenForClient(
  new string[]{ "https://management.azure.com//.default"})
   .ExecuteAsync();
ServiceClientCredentials cred = new TokenCredentials(result.AccessToken);
var client = new DataFactoryManagementClient(cred) {
    SubscriptionId = subscriptionId };

Tworzenie fabryki danych

Dodaj do metody Main poniższy kod, który tworzy fabrykę danych.

// Create a data factory
Console.WriteLine("Creating data factory " + dataFactoryName + "...");
Factory dataFactory = new Factory
{
    Location = region,
    Identity = new FactoryIdentity()
};
client.Factories.CreateOrUpdate(resourceGroup, dataFactoryName, dataFactory);
Console.WriteLine(
    SafeJsonConvert.SerializeObject(dataFactory, client.SerializationSettings));

while (client.Factories.Get(resourceGroup, dataFactoryName).ProvisioningState ==
       "PendingCreation")
{
    System.Threading.Thread.Sleep(1000);
}

Tworzenie usługi połączonej

Dodaj do metody Main poniższy kod, który tworzy połączoną usługę Azure Storage.

Połączone usługi tworzy się w fabryce danych w celu połączenia magazynów danych i usług obliczeniowych z fabryką danych. W tym przewodniku Szybki start musisz utworzyć tylko jedną połączoną usługę Azure Storage zarówno dla źródła kopii, jak i magazynu ujścia; nazwa to "AzureStorageLinkedService" w przykładzie.

// Create an Azure Storage linked service
Console.WriteLine("Creating linked service " + storageLinkedServiceName + "...");

LinkedServiceResource storageLinkedService = new LinkedServiceResource(
    new AzureStorageLinkedService
    {
        ConnectionString = new SecureString(
            "DefaultEndpointsProtocol=https;AccountName=" + storageAccount +
            ";AccountKey=" + storageKey)
    }
);
client.LinkedServices.CreateOrUpdate(
    resourceGroup, dataFactoryName, storageLinkedServiceName, storageLinkedService);
Console.WriteLine(SafeJsonConvert.SerializeObject(
    storageLinkedService, client.SerializationSettings));

Utwórz zestaw danych

Dodaj następujący kod do metody Main , która tworzy zestaw danych obiektów blob platformy Azure.

Zdefiniuj zestaw danych, który reprezentuje dane do skopiowania ze źródła do ujścia. W tym przykładzie ten zestaw danych obiektu blob odwołuje się do połączonej usługi Azure Storage utworzonej w poprzednim kroku. Zestaw danych przyjmuje parametr, którego wartość jest ustawiana w działaniu wykorzystującym zestaw danych. Parametr służy do konstruowania ścieżki "folderPath" wskazującej miejsce przechowywania/przechowywania danych.

// Create an Azure Blob dataset
Console.WriteLine("Creating dataset " + blobDatasetName + "...");
DatasetResource blobDataset = new DatasetResource(
    new AzureBlobDataset
    {
        LinkedServiceName = new LinkedServiceReference
        {
            ReferenceName = storageLinkedServiceName
        },
        FolderPath = new Expression { Value = "@{dataset().path}" },
        Parameters = new Dictionary<string, ParameterSpecification>
        {
            { "path", new ParameterSpecification { Type = ParameterType.String } }
        }
    }
);
client.Datasets.CreateOrUpdate(
    resourceGroup, dataFactoryName, blobDatasetName, blobDataset);
Console.WriteLine(
    SafeJsonConvert.SerializeObject(blobDataset, client.SerializationSettings));

Tworzenie potoku

Dodaj do metody Main poniższy kod, który tworzy potok z działaniem kopiowania.

W tym przykładzie ten potok zawiera jedno działanie i przyjmuje dwa parametry: ścieżkę wejściowego obiektu blob i ścieżkę wyjściowego obiektu blob. Wartości tych parametrów są ustawiane w chwili wyzwolenia/uruchomienia potoku. Działanie kopiowania dotyczy tego samego zestawu danych obiektu blob, który został utworzony w poprzednim kroku jako wejście i wyjście. W przypadku użycia zestawu danych jako zestawu danych wejściowych określana jest ścieżka wejściowa. Natomiast w przypadku użycia zestawu danych jako zestawu danych wyjściowych określana jest ścieżka wyjściowa.

// Create a pipeline with a copy activity
Console.WriteLine("Creating pipeline " + pipelineName + "...");
PipelineResource pipeline = new PipelineResource
{
    Parameters = new Dictionary<string, ParameterSpecification>
    {
        { "inputPath", new ParameterSpecification { Type = ParameterType.String } },
        { "outputPath", new ParameterSpecification { Type = ParameterType.String } }
    },
    Activities = new List<Activity>
    {
        new CopyActivity
        {
            Name = "CopyFromBlobToBlob",
            Inputs = new List<DatasetReference>
            {
                new DatasetReference()
                {
                    ReferenceName = blobDatasetName,
                    Parameters = new Dictionary<string, object>
                    {
                        { "path", "@pipeline().parameters.inputPath" }
                    }
                }
            },
            Outputs = new List<DatasetReference>
            {
                new DatasetReference
                {
                    ReferenceName = blobDatasetName,
                    Parameters = new Dictionary<string, object>
                    {
                        { "path", "@pipeline().parameters.outputPath" }
                    }
                }
            },
            Source = new BlobSource { },
            Sink = new BlobSink { }
        }
    }
};
client.Pipelines.CreateOrUpdate(resourceGroup, dataFactoryName, pipelineName, pipeline);
Console.WriteLine(SafeJsonConvert.SerializeObject(pipeline, client.SerializationSettings));

Tworzenie uruchomienia potoku

Dodaj do metody Main poniższy kod, który wyzwala uruchomienie potoku.

Ten kod ustawia również wartości parametrów inputPath i outputPath określonych w potoku przy użyciu rzeczywistych wartości ścieżek obiektu blob źródła i ujścia.

// Create a pipeline run
Console.WriteLine("Creating pipeline run...");
Dictionary<string, object> parameters = new Dictionary<string, object>
{
    { "inputPath", inputBlobPath },
    { "outputPath", outputBlobPath }
};
CreateRunResponse runResponse = client.Pipelines.CreateRunWithHttpMessagesAsync(
    resourceGroup, dataFactoryName, pipelineName, parameters: parameters
).Result.Body;
Console.WriteLine("Pipeline run ID: " + runResponse.RunId);

Monitorowanie uruchomienia potoku

Dodaj do metody Main poniższy kod, aby stale sprawdzać stan do momentu zakończenia kopiowania danych.

// Monitor the pipeline run
Console.WriteLine("Checking pipeline run status...");
PipelineRun pipelineRun;
while (true)
{
    pipelineRun = client.PipelineRuns.Get(
        resourceGroup, dataFactoryName, runResponse.RunId);
    Console.WriteLine("Status: " + pipelineRun.Status);
    if (pipelineRun.Status == "InProgress" || pipelineRun.Status == "Queued")
        System.Threading.Thread.Sleep(15000);
    else
        break;
}

Dodaj następujący kod do metody Main , która pobiera szczegóły przebiegu działania kopiowania, takie jak rozmiar odczytanych lub zapisanych danych.

// Check the copy activity run details
Console.WriteLine("Checking copy activity run details...");

RunFilterParameters filterParams = new RunFilterParameters(
    DateTime.UtcNow.AddMinutes(-10), DateTime.UtcNow.AddMinutes(10));
ActivityRunsQueryResponse queryResponse = client.ActivityRuns.QueryByPipelineRun(
    resourceGroup, dataFactoryName, runResponse.RunId, filterParams);
if (pipelineRun.Status == "Succeeded")
    Console.WriteLine(queryResponse.Value.First().Output);
else
    Console.WriteLine(queryResponse.Value.First().Error);
Console.WriteLine("\nPress any key to exit...");
Console.ReadKey();

Uruchamianie kodu

Skompiluj i uruchom aplikację, a następnie zweryfikuj wykonywanie potoku.

Konsola wypisuje postęp tworzenia fabryki danych, połączonej usługi, zestawów danych, potoku i uruchomienia potoku. Następnie sprawdza stan uruchomienia potoku. Poczekaj, aż zobaczysz szczegóły przebiegu działania kopiowania z rozmiarem danych odczytu/zapisu. Następnie użyj narzędzi, takich jak Eksplorator usługi Azure Storage, aby sprawdzić, czy obiekty blob są kopiowane do "outputBlobPath" z "inputBlobPath", jak określono w zmiennych.

Przykładowe dane wyjściowe

Creating data factory SPv2Factory0907...
{
  "identity": {
    "type": "SystemAssigned"
  },
  "location": "East US"
}
Creating linked service AzureStorageLinkedService...
{
  "properties": {
    "type": "AzureStorage",
    "typeProperties": {
      "connectionString": {
        "value": "DefaultEndpointsProtocol=https;AccountName=<storageAccountName>;AccountKey=<storageAccountKey>",
        "type": "SecureString"
      }
    }
  }
}
Creating dataset BlobDataset...
{
  "properties": {
    "type": "AzureBlob",
    "typeProperties": {
      "folderPath": {
        "value": "@{dataset().path}",
        "type": "Expression"
      }
    },
    "linkedServiceName": {
      "referenceName": "AzureStorageLinkedService",
      "type": "LinkedServiceReference"
    },
    "parameters": {
      "path": {
        "type": "String"
      }
    }
  }
}
Creating pipeline Adfv2QuickStartPipeline...
{
  "properties": {
    "activities": [
      {
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "BlobSink"
          }
        },
        "inputs": [
          {
            "referenceName": "BlobDataset",
            "parameters": {
              "path": "@pipeline().parameters.inputPath"
            },
            "type": "DatasetReference"
          }
        ],
        "outputs": [
          {
            "referenceName": "BlobDataset",
            "parameters": {
              "path": "@pipeline().parameters.outputPath"
            },
            "type": "DatasetReference"
          }
        ],
        "name": "CopyFromBlobToBlob"
      }
    ],
    "parameters": {
      "inputPath": {
        "type": "String"
      },
      "outputPath": {
        "type": "String"
      }
    }
  }
}
Creating pipeline run...
Pipeline run ID: 308d222d-3858-48b1-9e66-acd921feaa09
Checking pipeline run status...
Status: InProgress
Status: InProgress
Checking copy activity run details...
{
    "dataRead": 331452208,
    "dataWritten": 331452208,
    "copyDuration": 23,
    "throughput": 14073.209,
    "errors": [],
    "effectiveIntegrationRuntime": "DefaultIntegrationRuntime (West US)",
    "usedDataIntegrationUnits": 2,
    "billedDuration": 23
}

Press any key to exit...

Sprawdzanie danych wyjściowych

Potok automatycznie tworzy folder wyjściowy w kontenerze adftutorial blob. Następnie kopiuje plik emp.txt z folderu wejściowego do folderu wyjściowego.

W Azure Portal na stronie kontenera adftutorial zatrzymanego w folderze Dodaj folder wejściowy i plik dla powyższego kontenera obiektów blob wybierz pozycję Odśwież, aby wyświetlić folder wyjściowy.
Na liście folderów wybierz pozycję Dane wyjściowe.
Upewnij się, że plik emp.txt jest kopiowany do folderu wyjściowego.

Czyszczenie zasobów

Aby programowo usunąć fabrykę danych, dodaj do programu następujące wiersze kodu:

Console.WriteLine("Deleting the data factory");
client.Factories.Delete(resourceGroup, dataFactoryName);

Następne kroki

Potok w tym przykładzie kopiuje dane z jednej lokalizacji do innej lokalizacji w usłudze Azure Blob Storage. Zapoznaj się z samouczkami, aby dowiedzieć się więcej o korzystaniu z usługi Data Factory w dalszych scenariuszach.