Dela via

Snabbstart: Skapa en datafabrik och pipeline med .NET SDK

  • Artikel

GÄLLER FÖR: Azure Data Factory Azure Synapse Analytics

Dricks

Prova Data Factory i Microsoft Fabric, en allt-i-ett-analyslösning för företag. Microsoft Fabric omfattar allt från dataflytt till datavetenskap, realtidsanalys, business intelligence och rapportering. Lär dig hur du startar en ny utvärderingsversion kostnadsfritt!

Den här snabbstarten beskriver hur du använder .NET SDK för att skapa en Azure Data Factory. Den pipeline du skapar i den här datafabriken kopierar data från en mapp till en annan mapp i Azure Blob Storage. Om du vill se en självstudie som visar hur du omvandlar data med Azure Data Factory går du till Tutorial: Transform data using Spark (Självstudie: Omvandla data med Spark).

Kommentar

Den här artikeln ger inte någon detaljerad introduktion till Azure Data Factory-tjänsten. En introduktion till Azure Data Factory-tjänsten finns i Introduktion till Azure Data Factory.

Förutsättningar

Azure-prenumeration

Om du inte har någon Azure-prenumeration skapar du ett kostnadsfritt konto innan du börjar.

Azure-roller

Om du vill skapa Data Factory-instanser måste det användarkonto du använder för att logga in på Azure vara medlem av rollerna deltagare eller ägare, eller vara administratör för Azure-prenumerationen. Om du vill visa de behörigheter som du har i prenumerationen går du till Azure-portalen, väljer ditt användarnamn i det övre högra hörnet, väljer ikonen "..." för fler alternativ och väljer sedan Mina behörigheter. Om du har åtkomst till flera prenumerationer väljer du rätt prenumeration.

För att skapa och hantera underordnade resurser för Data Factory – inklusive datauppsättningar, länkade tjänster, pipelines, utlösare och integreringskörningar – gäller följande krav:

  • För att kunna skapa och hantera underordnade resurser i Azure-portalen måste du tillhöra rollen Data Factory-deltagare på resursgruppsnivå eller högre.
  • För att skapa och hantera underordnade resurser med PowerShell eller SDK räcker det att du har rollen som deltagare på resursnivå eller högre.

För exempel på instruktioner om hur du lägger till en användare till en roll läser du artikeln Lägg till roller.

Mer information finns i följande artiklar:

Azure-lagringskonto

Du använder ett Allmänt Azure Storage-konto (särskilt Blob Storage) som både käll - och måldatalager i den här snabbstarten. Om du inte har något allmänt Azure Storage-konto kan du läsa Skapa ett lagringskonto för att skapa ett.

Hämta namnet på lagringskontot

Du behöver namnet på ditt Azure Storage-konto för den här snabbstarten. Följande procedur innehåller steg för att hämta namnet på ditt lagringskonto:

  1. I en webbläsare går du till Azure-portalen och loggar in med ditt Användarnamn och lösenord för Azure.
  2. På Menyn i Azure-portalen väljer du Alla tjänster och sedan Lagringslagringskonton>. Du kan också söka efter och välja Lagringskonton från valfri sida.
  3. På sidan Lagringskonton filtrerar du efter ditt lagringskonto (om det behövs) och väljer sedan ditt lagringskonto.

Du kan också söka efter och välja Lagringskonton från valfri sida.

Skapa en blobcontainer

I det här avsnittet skapar du en blobcontainer med namnet adftutorial i Azure Blob Storage.

  1. På sidan lagringskonto väljer du Översiktscontainrar>.

  2. I verktygsfältet För kontonamncontainrar - > väljer du Container.<

  3. I dialogrutan Ny container anger du adftutorial som namn och väljer OK. Sidan Containrar för kontonamn - >uppdateras så att den innehåller adftutorial i listan över containrar.<

    Lista över behållare

Lägga till en indatamapp och fil för blobcontainern

I det här avsnittet skapar du en mapp med namnet input i containern som du skapade och laddar sedan upp en exempelfil till indatamappen. Innan du börjar öppnar du en textredigerare, till exempel Anteckningar, och skapar en fil med namnet emp.txt med följande innehåll:

John, Doe
Jane, Doe

Spara filen i mappen C:\ADFv2QuickStartPSH . (Om mappen inte redan finns skapar du den.) Gå sedan tillbaka till Azure-portalen och följ dessa steg:

  1. På sidan Kontonamncontainrar - > där du slutade väljer du adftutorial i den uppdaterade listan med containrar.<

    1. Om du stängde fönstret eller gick till en annan sida loggar du in på Azure-portalen igen.
    2. På Menyn i Azure-portalen väljer du Alla tjänster och sedan Lagringslagringskonton>. Du kan också söka efter och välja Lagringskonton från valfri sida.
    3. Välj ditt lagringskonto och välj sedan Containrar>adftutorial.

  2. I verktygsfältet för containersidan adftutorial väljer du Ladda upp.

  3. På sidan Ladda upp blob väljer du rutan Filer och bläddrar sedan till och väljer filen emp.txt .

  4. Expandera rubriken Avancerat . Sidan visas nu som den visas:

    Välj länken Avancerat

  5. I rutan Ladda upp till mapp anger du indata.

  6. Välj knappen Ladda upp. Du bör se filen emp.txt och uppladdningens status i listan.

  7. Välj ikonen Stäng (ett X) för att stänga sidan Ladda upp blob.

Håll sidan adftutorial container öppen. Du kommer att använda den för att bekräfta utdata i slutet av snabbstarten.

Visual Studio

Genomgången i den här artikeln använder Visual Studio 2019. Procedurerna för Visual Studio 2013, 2015 eller 2017 skiljer sig något åt.

Skapa ett program i Microsoft Entra-ID

I avsnitten i Så här: Använd portalen för att skapa ett Microsoft Entra-program och tjänstens huvudnamn som kan komma åt resurser följer du anvisningarna för att utföra följande uppgifter:

  1. I Skapa ett Microsoft Entra-program skapar du ett program som representerar det .NET-program som du skapar i den här självstudien. För inloggnings-URL kan du ange en låtsas-URL enligt artikeln (https://contoso.org/exampleapp).
  2. I Hämta värden för inloggning hämtar du program-ID och klientorganisations-ID och noterar dessa värden som du använder senare i den här självstudien.
  3. I Certifikat och hemligheter hämtar du autentiseringsnyckeln och noterar det här värdet som du använder senare i den här självstudien.
  4. I Tilldela programmet till en roll tilldelar du programmet rollen Deltagare på prenumerationsnivå så att programmet kan skapa datafabriker i prenumerationen.

Skapa ett Visual Studio-projekt

Skapa sedan ett C# .NET-konsolprogram i Visual Studio:

  1. Starta Visual Studio.
  2. I startfönstret väljer du Skapa ett nytt projekt>Konsolapp (.NET Framework). .NET version 4.5.2 eller senare krävs.
  3. I Projektnamn anger du ADFv2QuickStart.
  4. Välj Skapa för att skapa projektet.

Installera NuGet-paket

  1. Välj Verktyg>NuGet Upravljač za pakete> Upravljač za pakete Console.

  2. I fönstret Upravljač za pakete Konsol kör du följande kommandon för att installera paket. Mer information finns i Azure.ResourceManager.DataFactory NuGet-paketet.

    Install-Package Azure.ResourceManager.DataFactory -IncludePrerelease
Install-Package Azure.Identity

Skapa en datafabrik

  1. Öppna Program.cs, lägg till följande instruktioner för att lägga till referenser till namnområden.

    using Azure;
using Azure.Core;
using Azure.Core.Expressions.DataFactory;
using Azure.Identity;
using Azure.ResourceManager;
using Azure.ResourceManager.DataFactory;
using Azure.ResourceManager.DataFactory.Models;
using Azure.ResourceManager.Resources;
using System;
using System.Collections.Generic;

  2. Lägg till följande kod till Main-metoden som anger variablerna. Ersätt platshållarna med dina egna värden. Om du vill se en lista med Azure-regioner där Data Factory är tillgängligt för närvarande markerar du de regioner du är intresserad av på följande sida. Expandera sedan Analytics och leta rätt på Data Factory: Tillgängliga produkter per region. Datalager (Azure Storage, Azure SQL Database med mera) och beräkningar (HDInsight med flera) som används av datafabriken kan finnas i andra regioner.

    // 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 storageAccountName = "<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 inputBlobContainer = "<blob container to copy data from, e.g. containername>";
string inputBlobPath = "<path to existing blob(s) to copy data from, e.g. inputdir/file>";
//specify the contains and output folder where the files are copied
string outputBlobContainer = "<blob container to copy data from, e.g. containername>";
string outputBlobPath = "<the blob path to copy data to, e.g. outputdir/file>";

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

  3. Lägg till följande kod som skapar en datafabrik till Main-metoden.

    ArmClient armClient = new ArmClient(
    new ClientSecretCredential(tenantID, applicationId, authenticationKey, new TokenCredentialOptions
    {
        AuthorityHost = AzureAuthorityHosts.AzurePublicCloud
    }), 
    subscriptionId, 
    new ArmClientOptions { Environment = ArmEnvironment.AzurePublicCloud }
);

ResourceIdentifier resourceIdentifier = SubscriptionResource.CreateResourceIdentifier(subscriptionId);
SubscriptionResource subscriptionResource = armClient.GetSubscriptionResource(resourceIdentifier);

Console.WriteLine("Get an existing resource group " + resourceGroupName + "...");
var resourceGroupOperation = subscriptionResource.GetResourceGroups().Get(resourceGroupName);
ResourceGroupResource resourceGroupResource = resourceGroupOperation.Value;

Console.WriteLine("Create a data factory " + dataFactoryName + "...");
DataFactoryData dataFactoryData = new DataFactoryData(AzureLocation.EastUS2);
var dataFactoryOperation = resourceGroupResource.GetDataFactories().CreateOrUpdate(WaitUntil.Completed, dataFactoryName, dataFactoryData);
Console.WriteLine(dataFactoryOperation.WaitForCompletionResponse().Content);

// Get the data factory resource
DataFactoryResource dataFactoryResource = dataFactoryOperation.Value;

Skapa en länkad tjänst

Lägg till följande kod i Main-metoden som skapar en länkad Azure Storage-tjänst.

Du kan skapa länkade tjänster i en datafabrik för att länka ditt datalager och beräkna datafabrik-tjänster. I den här snabbstarten behöver du bara skapa en länkad Azure Blob Storage-tjänst för både kopieringskällan och mottagararkivet. den heter "AzureBlobStorageLinkedService" i exemplet.

// Create an Azure Storage linked service
Console.WriteLine("Create a linked service " + storageLinkedServiceName + "...");
AzureBlobStorageLinkedService azureBlobStorage = new AzureBlobStorageLinkedService()
{
    ConnectionString = azureBlobStorageConnectionString
};

DataFactoryLinkedServiceData linkedServiceData = new DataFactoryLinkedServiceData(azureBlobStorage);

var linkedServiceOperation = dataFactoryResource.GetDataFactoryLinkedServices().CreateOrUpdate(WaitUntil.Completed, storageLinkedServiceName, linkedServiceData);
Console.WriteLine(linkedServiceOperation.WaitForCompletionResponse().Content);

Skapa en datamängd

Lägg till följande kod i main-metoden som skapar en avgränsad textdatauppsättning.

Du definierar en datauppsättning som representerar data som ska kopieras från en källa till en mottagare. I det här exemplet refererar den här avgränsade textdatauppsättningen till den länkade Azure Blob Storage-tjänsten som du skapade i föregående steg. Datauppsättningen tar två parametrar vars värde anges i en aktivitet som använder datauppsättningen. Parametrarna används för att konstruera "containern" och "folderPath" som pekar på var data finns/lagras.

// Create an Azure Blob dataset
DataFactoryLinkedServiceReference linkedServiceReference = new DataFactoryLinkedServiceReference(DataFactoryLinkedServiceReferenceType.LinkedServiceReference, storageLinkedServiceName);
DelimitedTextDataset delimitedTextDataset = new DelimitedTextDataset(linkedServiceReference)
{
    DataLocation = new AzureBlobStorageLocation
    {
        Container = DataFactoryElement<string>.FromExpression("@dataset().container"),
        FileName = DataFactoryElement<string>.FromExpression("@dataset().path")
    },
    Parameters =
    {
        new KeyValuePair<string, EntityParameterSpecification>("container",new EntityParameterSpecification(EntityParameterType.String)),
        new KeyValuePair<string, EntityParameterSpecification>("path",new EntityParameterSpecification(EntityParameterType.String))
    },
    FirstRowAsHeader = false,
    QuoteChar = "\"",
    EscapeChar = "\\",
    ColumnDelimiter = ","
};

DataFactoryDatasetData datasetData = new DataFactoryDatasetData(delimitedTextDataset);

var datasetOperation = dataFactoryResource.GetDataFactoryDatasets().CreateOrUpdate(WaitUntil.Completed, blobDatasetName, datasetData);
Console.WriteLine(datasetOperation.WaitForCompletionResponse().Content);

Skapa en pipeline

Lägg till följande kod som skapar och aktiverar en pipeline till Main-metoden.

I det här exemplet innehåller den här pipelinen en aktivitet och tar fyra parametrar: indatablobcontainern och sökvägen samt utdatablobcontainern och sökvägen. Värdena för dessa parametrar anges när pipeline utlöses/körs. Kopieringsaktiviteten refererar till samma blobdatauppsättning som skapats i föregående steg som indata och utdata. När datamängden används som indatauppsättning anges indatacontainer och sökväg. Och när datamängden används som en utdatauppsättning anges utdatacontainern och sökvägen.

// Create a pipeline with a copy activity
Console.WriteLine("Creating pipeline " + pipelineName + "...");
DataFactoryPipelineData pipelineData = new DataFactoryPipelineData()
{
    Parameters =
    {
        new KeyValuePair<string, EntityParameterSpecification>("inputContainer",new EntityParameterSpecification(EntityParameterType.String)),
        new KeyValuePair<string, EntityParameterSpecification>("inputPath",new EntityParameterSpecification(EntityParameterType.String)),
        new KeyValuePair<string, EntityParameterSpecification>("outputContainer",new EntityParameterSpecification(EntityParameterType.String)),
        new KeyValuePair<string, EntityParameterSpecification>("outputPath",new EntityParameterSpecification(EntityParameterType.String))
    },
    Activities =
    {
        new CopyActivity("CopyFromBlobToBlob",new DataFactoryBlobSource(),new DataFactoryBlobSink())
        {
            Inputs =
            {
                new DatasetReference(DatasetReferenceType.DatasetReference,blobDatasetName)
                {
                    Parameters =
                    {
                        new KeyValuePair<string, BinaryData>("container", BinaryData.FromString("\"@pipeline().parameters.inputContainer\"")),
                        new KeyValuePair<string, BinaryData>("path", BinaryData.FromString("\"@pipeline().parameters.inputPath\""))
                    }
                }
            },
            Outputs =
            {
                new DatasetReference(DatasetReferenceType.DatasetReference,blobDatasetName)
                {
                    Parameters =
                    {
                        new KeyValuePair<string, BinaryData>("container", BinaryData.FromString("\"@pipeline().parameters.outputContainer\"")),
                        new KeyValuePair<string, BinaryData>("path", BinaryData.FromString("\"@pipeline().parameters.outputPath\""))
                    }
                }
            }
        }
    }
};

var pipelineOperation = dataFactoryResource.GetDataFactoryPipelines().CreateOrUpdate(WaitUntil.Completed, pipelineName, pipelineData);
Console.WriteLine(pipelineOperation.WaitForCompletionResponse().Content);

Skapa en pipelinekörning

Lägg till följande kod i Main-metoden som utlöser en pipelinekörning.

Den här koden anger även värden för parametrarna inputContainer, inputPath, outputContainer och outputPath som anges i pipelinen med de faktiska värdena för käll- och mottagarblobsökvägarna.

// Create a pipeline run
Console.WriteLine("Creating pipeline run...");
Dictionary<string, BinaryData> parameters = new Dictionary<string, BinaryData>()
{
    { "inputContainer",BinaryData.FromObjectAsJson(inputBlobContainer) },
    { "inputPath",BinaryData.FromObjectAsJson(inputBlobPath) },
    { "outputContainer",BinaryData.FromObjectAsJson(outputBlobContainer) },
    { "outputPath",BinaryData.FromObjectAsJson(outputBlobPath) }
};

var pipelineResource = dataFactoryResource.GetDataFactoryPipeline(pipelineName);
var runResponse = pipelineResource.Value.CreateRun(parameters);
Console.WriteLine("Pipeline run ID: " + runResponse.Value.RunId);

Övervaka en pipelinekörning

  1. Lägg till följande kod i Main-metoden för att kontinuerligt kontrollera status tills den har slutat att kopiera data.

    // Monitor the pipeline run
Console.WriteLine("Checking pipeline run status...");
DataFactoryPipelineRunInfo pipelineRun;
while (true)
{
    pipelineRun = dataFactoryResource.GetPipelineRun(runResponse.Value.RunId.ToString());
    Console.WriteLine("Status: " + pipelineRun.Status);
    if (pipelineRun.Status == "InProgress" || pipelineRun.Status == "Queued")
        System.Threading.Thread.Sleep(15000);
    else
        break;
}

  2. Lägg till följande kod i main-metoden som hämtar information om kopieringsaktivitetskörning, till exempel storleken på de data som läs- eller skrivs.

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

var queryResponse = dataFactoryResource.GetActivityRun(pipelineRun.RunId.ToString(), 
    new RunFilterContent(DateTime.UtcNow.AddMinutes(-10), DateTime.UtcNow.AddMinutes(10)));

var enumerator = queryResponse.GetEnumerator();
enumerator.MoveNext();

if (pipelineRun.Status == "Succeeded")
    Console.WriteLine(enumerator.Current.Output);
else
    Console.WriteLine(enumerator.Current.Error);
Console.WriteLine("\nPress any key to exit...");
Console.ReadKey();

Kör koden

Skapa och starta programmet och kontrollera sedan pipelinekörningen.

Konsolen skriver ut förloppet för skapandet av datafabriken, den länkade tjänsten, datauppsättningar, pipeline och pipelinekörning. Sedan kontrolleras status för pipelinekörningen. Vänta tills du ser information om kopieringsaktivitetens körning med storleken på läs-/skrivdata. Använd sedan verktyg som Azure Storage Explorer för att kontrollera att blobarna kopieras till "outputBlobPath" från "inputBlobPath" som du angav i variablerna.

Exempelutdata

Create a data factory quickstart-adf...
{
  "name": "quickstart-adf",
  "type": "Microsoft.DataFactory/factories",
  "properties": {
    "provisioningState": "Succeeded",
    "version": "2018-06-01"
  },
  "location": "eastus2"
}
Create a linked service AzureBlobStorage...
{
  "name": "AzureBlobStorage",
  "type": "Microsoft.DataFactory/factories/linkedservices",
  "properties": {
    "type": "AzureBlobStorage",
    "typeProperties": {
      "connectionString": "DefaultEndpointsProtocol=https;AccountName=<storageAccountName>;",
      "encryptedCredential": "<encryptedCredential>"
    }
  }
}
Creating dataset BlobDelimitedDataset...
{
  "name": "BlobDelimitedDataset",
  "type": "Microsoft.DataFactory/factories/datasets",
  "properties": {
    "type": "DelimitedText",
    "linkedServiceName": {
      "type": "LinkedServiceReference",
      "referenceName": "AzureBlobStorage"
    },
    "parameters": {
      "container": {
        "type": "String"
      },
      "path": {
        "type": "String"
      }
    },
    "typeProperties": {
      "location": {
        "container": {
          "type": "Expression",
          "value": "@dataset().container"
        },
        "type": "AzureBlobStorageLocation",
        "fileName": {
          "type": "Expression",
          "value": "@dataset().path"
        }
      },
      "columnDelimiter": ",",
      "quoteChar": "\"",
      "escapeChar": "\\",
      "firstRowAsHeader": false
    }
  }
}
Creating pipeline Adfv2QuickStartPipeline...
{
  "properties": {
    "activities": [
      {
        "inputs": [
          {
            "type": "DatasetReference",
            "referenceName": "BlobDelimitedDataset",
            "parameters": {
              "container": "@pipeline().parameters.inputContainer",
              "path": "@pipeline().parameters.inputPath"
            }
          }
        ],
        "outputs": [
          {
            "type": "DatasetReference",
            "referenceName": "BlobDelimitedDataset",
            "parameters": {
              "container": "@pipeline().parameters.outputContainer",
              "path": "@pipeline().parameters.outputPath"
            }
          }
        ],
        "name": "CopyFromBlobToBlob",
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "BlobSink"
          }
        }
      }
    ],
    "parameters": {
      "inputContainer": {
        "type": "String"
      },
      "inputPath": {
        "type": "String"
      },
      "outputContainer": {
        "type": "String"
      },
      "outputPath": {
        "type": "String"
      }
    }
  }
}
Creating pipeline run...
Pipeline run ID: 3aa26ffc-5bee-4db9-8bac-ccbc2d7b51c1
Checking pipeline run status...
Status: InProgress
Status: Succeeded
Checking copy activity run details...
{
  "dataRead": 1048,
  "dataWritten": 1048,
  "filesRead": 1,
  "filesWritten": 1,
  "sourcePeakConnections": 1,
  "sinkPeakConnections": 1,
  "copyDuration": 8,
  "throughput": 1.048,
  "errors": [],
  "effectiveIntegrationRuntime": "AutoResolveIntegrationRuntime (East US 2)",
  "usedDataIntegrationUnits": 4,
  "billingReference": {
    "activityType": "DataMovement",
    "billableDuration": [
      {
        "meterType": "AzureIR",
        "duration": 0.06666666666666667,
        "unit": "DIUHours"
      }
    ],
    "totalBillableDuration": [
      {
        "meterType": "AzureIR",
        "duration": 0.06666666666666667,
        "unit": "DIUHours"
      }
    ]
  },
  "usedParallelCopies": 1,
  "executionDetails": [
    {
      "source": {
        "type": "AzureBlobStorage"
      },
      "sink": {
        "type": "AzureBlobStorage"
      },
      "status": "Succeeded",
      "start": "2023-12-15T10:25:33.9991558Z",
      "duration": 8,
      "usedDataIntegrationUnits": 4,
      "usedParallelCopies": 1,
      "profile": {
        "queue": {
          "status": "Completed",
          "duration": 5
        },
        "transfer": {
          "status": "Completed",
          "duration": 1,
          "details": {
            "listingSource": {
              "type": "AzureBlobStorage",
              "workingDuration": 0
            },
            "readingFromSource": {
              "type": "AzureBlobStorage",
              "workingDuration": 0
            },
            "writingToSink": {
              "type": "AzureBlobStorage",
              "workingDuration": 0
            }
          }
        }
      },
      "detailedDurations": {
        "queuingDuration": 5,
        "transferDuration": 1
      }
    }
  ],
  "dataConsistencyVerification": {
    "VerificationResult": "NotVerified"
  }
}

Press any key to exit...

Verifiera utdata

Pipelinen skapar automatiskt utdatamappen i blobcontainern adftutorial . Sedan kopieras filen emp.txt från indatamappen till utdatamappen.

  1. I Azure-portalen går du till sidan adftutorial container som du stannade vid i avsnittet Lägg till en indatamapp och fil för blobcontainern ovan och väljer Uppdatera för att se utdatamappen.
  2. I mapplistan väljer du utdata.
  3. Kontrollera att emp.txt har kopierats till utdatamappen.

Rensa resurser

Om du vill ta bort datafabriken programmatiskt lägger du till följande kodrader i programmet:

Console.WriteLine("Deleting the data factory");
dataFactoryResource.Delete(WaitUntil.Completed);

Nästa steg

Pipeline i det här exemplet kopierar data från en plats till en annan i Azure Blob Storage. Gå igenom självstudiekurserna om du vill lära dig hur du använder Data Factory i fler scenarier.