Mulai Cepat: Buat pabrik data dan alur menggunakan .NET SDK

Artikel
03/09/2023
11 menit untuk membaca

BERLAKU UNTUK: Azure Data Factory Azure Synapse Analytics

Mulai cepat ini menjelaskan cara menggunakan Azure CLI untuk membuat Azure Data Factory. Alur yang Anda buat di pabrik data ini menyalin data dari satu folder ke folder lain dalam penyimpanan blob Azure. Untuk tutorial tentang cara mengubah data menggunakan Azure Data Factory, lihat Tutorial: Mengubah data menggunakan Spark.

Catatan

Artikel ini tidak memberikan pengenalan mendetail tentang layanan Data Factory. Untuk pengenalan tentang layanan Azure Data Factory, lihat Pengantar Azure Data Factory.

Prasyarat

Langganan Azure

Jika Anda tidak memiliki langganan Azure, buat akun gratis sebelum Anda memulai.

Peran Azure

Untuk membuat instans Data Factory, akun pengguna yang Anda gunakan untuk masuk ke Azure harus menjadi anggota peran kontributor atau peran pemilik, atau administrator langganan Azure. Untuk menampilkan izin yang Anda miliki di langganan, buka portal Azure, pilih nama pengguna Anda di sudut kanan atas, pilih ikon " ... " untuk opsi selengkapnya, lalu pilih Izin saya. Jika Anda memiliki akses ke beberapa langganan, pilih langganan yang sesuai.

Untuk membuat dan mengelola sumber daya anak untuk Data Factory - termasuk himpunan data, layanan tertaut, alur, pemicu, dan runtime integrasi - persyaratan berikut ini berlaku:

Untuk membuat dan mengelola sumber daya anak di portal Microsoft Azure, Anda harus memiliki peran Kontributor Data Factory di tingkat Grup Sumber Daya atau di atasnya.
Untuk membuat dan mengelola sumber daya anak dengan PowerShell atau SDK, peran kontributor di tingkat sumber daya atau di atasnya cukup.

Untuk sampel instruksi tentang cara menambahkan pengguna ke peran, lihat artikel Menambahkan peran.

Untuk informasi selengkapnya, baca artikel berikut:

Akun Azure Storage

Anda menggunakan akun Azure Storage serba guna (khususnya penyimpanan Blob) sebagai sumber dan tujuan penyimpanan data dalam mulai cepat ini. Jika Anda belum punya akun penyimpanan Azure Storage serba guna, lihat Buat akun penyimpanan untuk membuatnya.

Dapatkan nama akun penyimpanan

Anda memerlukan nama akun Azure Storage Anda untuk mulai cepat ini. Prosedur berikut ini menyediakan langkah-langkah untuk mendapatkan nama akun penyimpanan Anda:

Di browser web, buka portal Azure, lalu masuk menggunakan nama pengguna dan kata sandi Azure Anda.
Dari menu portal Azure, pilih Semua layanan, lalu pilih Penyimpanan>Akun penyimpanan. Anda juga dapat mencari dan memilih Akun penyimpanan dari halaman mana pun.
Dalam halaman Akun penyimpanan, filter untuk akun penyimpanan Anda (jika diperlukan), lalu pilih akun penyimpanan Anda.

Anda juga dapat mencari dan memilih Akun penyimpanan dari halaman mana pun.

Membuat kontainer blob

Di bagian ini, Anda membuat kontainer blob bernama adftutorial di penyimpanan Azure Blob.

Dari halaman akun penyimpanan, pilih Gambaran Umum>Kontainer.
Pada bar alat halaman <Nama akun> - Kontainer, pilih Kontainer.
Di Kotak dialog Kontainer baru, masukkan adftutorial untuk nama, lalu pilih OK. halaman <Nama akun> - Kontainer diperbarui untuk menyertakan adftutorial dalam daftar kontainer.

Menambahkan folder input dan file untuk kontainer blob

Di bagian ini, Anda membuat folder bernama input dalam kontainer yang Anda buat, lalu mengunggah file sampel ke folder input. Sebelum memulai, buka editor teks seperti Notepad, dan buat file bernama emp.txt dengan konten berikut:

John, Doe
Jane, Doe

Simpan file di folder C:\ADFv2QuickStartPSH. (Jika folder belum ada, buatlah.) Lalu kembali ke portal Azure dan ikuti langkah-langkah berikut:

Di halaman <Nama akun> - Kontainer tempat yang Anda tinggalkan, pilih adftutorial dari daftar kontainer yang diperbarui.
1. Jika Anda menutup jendela atau pergi ke halaman lain, masuk lagi ke portal Azure.
2. Dari menu portal Azure, pilih Semua layanan, lalu pilih Penyimpanan>Akun penyimpanan. Anda juga dapat mencari dan memilih Akun penyimpanan dari halaman mana pun.
3. Pilih akun penyimpanan Anda, lalu pilih Kontainer>adftutorial.
Pada tollbar halaman kontainer adftutorial, pilih Unggah.
Di halaman Unggah blob, pilih kotak File, lalu telusuri dan pilih file emp.txt.
Memperluas judul Tingkat Lanjut. Halaman sekarang menampilkan seperti yang ditunjukkan:
Di kotak Unggah ke folder, masukkan input.
Pilih tombol Unggah. Anda harus melihat file emp.txt dan status unggahan dalam daftar.
Pilih ikon Tutup (X) untuk menutup halaman Unggah blob.

Jaga halaman kontainer adftutorial tetap terbuka. Anda menggunakannya untuk memverifikasi output di akhir mulai cepat ini.

Visual Studio

Panduan dalam artikel ini menggunakan Visual Studio 2019. Prosedur untuk Visual Studio 2013, 2015, atau 2017 sedikit berbeda.

Buat aplikasi di Azure Active Directory

Dari bagian di Cara kerja: Menggunakan portal untuk membuat aplikasi Azure Active Directory dan perwakilan layanan yang dapat mengakses sumber daya, ikuti petunjuk untuk melakukan tugas-tugas ini:

Di Buat aplikasi Azure Active Directory, buat aplikasi yang mewakili aplikasi .NET yang Anda buat dalam tutorial ini. Untuk URL masuk, Anda dapat memberikan URL dummy seperti yang ditunjukkan dalam artikel (https://contoso.org/exampleapp).
Di Dapatkan nilai untuk masuk, dapatkan ID aplikasi dan ID penyewa, dan catat nilai-nilai ini yang Anda gunakan nanti dalam tutorial ini.
Dalam Sertifikat dan rahasia, dapatkan kunci autentikasi, dan catat nilai ini yang Anda gunakan nanti dalam tutorial ini.
Di Tetapkan aplikasi ke peran, tetapkan aplikasi ke peran Kontributor pada tingkat langganan sehingga aplikasi dapat membuat pabrik data dalam langganan.

Membuat proyek Visual Studio

Selanjutnya, buat aplikasi konsol C# .NET di Visual Studio:

Luncurkan Visual Studio.
Di jendela Mulai, pilih Buat Aplikasi Konsol proyek baru>(.NET Framework) . Diperlukan .NET versi 4.5.2 atau lebih tinggi.
Di Nama proyek, masukkan ADFv2QuickStart.
Pilih Buat untuk membuat proyek.

Pasang paket NuGet

Pilih Alat>Manajer Paket NuGet>Konsol Manajer Paket.

Di Konsol Manajer Paket, jalankan perintah berikut untuk memasang paket. Untuk informasi selengkapnya, lihat paket NuGet Microsoft.Azure.Management.DataFactory.

Install-Package Microsoft.Azure.Management.DataFactory
Install-Package Microsoft.Azure.Management.ResourceManager -IncludePrerelease
Install-Package Microsoft.Identity.Client

Membuat klien pabrik data

Buka Program.cs, termasuk pernyataan berikut untuk menambahkan referensi ke kumpulan nama XML.

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;

Tambahkan kode berikut ke metode Utama yang mengatur variabel. Ganti tempat penampung dengan nilai Anda. Untuk daftar wilayah Azure tempat Data Factory saat ini tersedia, pilih wilayah yang menarik minat Anda pada halaman berikut, lalu perluas Analitik untuk menemukan Data Factory: Produk yang tersedia menurut wilayah. Penyimpanan data (Azure Storage, Azure SQL Database, dll.) dan komputasi (HDInsight, dll.) yang digunakan oleh pabrik data dapat berada di wilayah lain.

// 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";

Catatan

Untuk Sovereign cloud, Anda harus menggunakan titik akhir khusus cloud yang sesuai untuk ActiveDirectoryAuthority dan ResourceManagerUrl (BaseUri). Misalnya, di US Azure Gov Anda akan menggunakan https://login.microsoftonline.us otoritas daripada https://login.microsoftonline.com, dan menggunakan https://management.usgovcloudapi.net daripada https://management.azure.com/, dan kemudian membuat klien manajemen pabrik data. Anda dapat menggunakan Powershell guna mendapatkan Url titik akhir dengan mudah untuk berbagai cloud dengan menjalankan "Get-AzEnvironment | Format-List", yang akan menampilkan daftar titik akhir untuk setiap lingkungan cloud.

Tambahkan kode berikut ke metode Utama yang membuat instans kelas DataFactoryManagementClient. Anda menggunakan objek ini untuk membuat pabrik data, layanan tertaut, himpunan data, dan alur. Anda juga menggunakan obyek ini untuk memantau detail eksekusi alur.

// 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 };

Buat pabrik data

Tambahkan kode berikut ke metode Utama yang membuat pabrik data.

// 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);
}

Membuat layanan tertaut

Tambahkan kode berikut ke metode Utama yang membuat layanan tertaut Azure Storage.

Anda membuat layanan tertaut di pabrik data untuk menautkan penyimpanan data dan layanan komputasi ke pabrik data. Dalam mulai cepat ini, Anda hanya perlu membuat satu layanan tertaut Azure Storage sebagai sumber salinan dan penyimpanan sink, bernama "AzureStorageLinkedService" dalam sampel.

// 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));

Buat himpunan data

Tambahkan kode berikut ke metode Utama yang membuat himpunan data blob Azure.

Anda menentukan himpunan data yang mewakili data untuk disalin dari sumber ke sink. Dalam contoh ini, himpunan data Blob ini mengacu pada layanan tertaut Azure Storage yang telah Anda buat di langkah sebelumnya. Himpunan data mengambil parameter yang nilainya diatur pada aktivitas yang memerlukan himpunan data. Parameter digunakan untuk membangun "folderPath" yang menunjuk ke tempat data berada/ disimpan.

// 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));

Membuat alur

Tambahkan kode berikut ke metode Utama yang membuat alur dengan aktivitas salin.

Dalam contoh ini, alur berisi satu aktivitas dan mengambil dua parameter: jalur blob input dan jalur blob output. Nilai untuk parameter ini diatur saat alur dipicu/dijalankan. Aktivitas salin mengacu pada set data gumpalan yang sama yang dibuat pada langkah sebelumnya sebagai input dan output. Ketika himpunan data digunakan sebagai himpunan data input, jalur input ditentukan. Dan, ketika himpunan data digunakan sebagai himpunan data output, jalur output ditentukan.

// 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));

Membuat eksekusi alur

Tambahkan kode berikut ke metode Utama yang memicu eksekusi alur.

Kode ini juga menetapkan nilai parameter inputPath dan outputPath yang ditentukan di alur dengan nilai aktual dari jalur sumber dan blob sink.

// 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);

Memantau eksekusi alur

Tambahkan kode berikut ke metode Utama untuk terus memeriksa status hingga selesai menyalin data.

// 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;
}

Tambahkan kode berikut ke metode Utama yang mengambil salinan aktivitas menjalankan detail, seperti ukuran data yang dibaca atau ditulis.

// 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();

Jalankan kode

Bangun dan mulai aplikasi, lalu verifikasi eksekusi alur.

Konsol mencetak kemajuan pembuatan pabrik data, layanan tertaut, himpunan data, alur, dan eksekusi alur. Kemudian memeriksa status eksekusi alur. Tunggu hingga Anda melihat aktivitas salin sedang menjalankan detail dengan ukuran data baca/tulis. Kemudian, gunakan alat seperti Penjelajah Azure Storage untuk memeriksa blob disalin ke "outputBlobPath" dari "inputBlobPath" seperti yang Anda tentukan dalam variabel.

Sampel output

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

Verifikasi output

Alur secara otomatis membuat folder output di kontainer blob adftutorial. Kemudian, menyalin file emp.txt dari folder input ke folder output.

Di portal Microsoft Azure, pada halaman kontainer adftutorial tempat Anda berhenti di bagian Tambahkan folder input dan file untuk kontainer blob di atas, pilih Refresh untuk melihat folder output.
Pada daftar folder, pilih output.
Pastikan bahwa emp.txt disalin ke folder output.

Membersihkan sumber daya

Untuk menghapus pabrik data secara terprogram, tambahkan baris kode berikut ke program:

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

Langkah berikutnya

Alur dalam sampel ini menyalin data dari satu lokasi ke lokasi lain dalam penyimpanan blob Azure. Ikuti tutorial untuk mempelajari tentang penggunaan Data Factory dalam skenario lainnya.