Data Factory パイプラインでのアクティビティの分岐と連鎖

[アーティクル]
10/03/2024

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータサイエンス、リアルタイム分析、ビジネスインテリジェンス、レポートまで、あらゆるものをカバーしています。無料で新しい試用版を開始する方法について説明します。

このチュートリアルでは、いくつかの制御フロー機能を紹介する Data Factory パイプラインを作成します。このパイプラインでは、Azure Blob Storage 内のコンテナーから同じストレージアカウント内の別のコンテナーへのコピーを行います。コピーアクティビティが成功した場合は、成功したコピー操作の詳細がパイプラインによってメールで送信されます。この情報には、書き込まれたデータの量が含まれる場合があります。コピーアクティビティが失敗した場合は、エラーメッセージなどのコピー失敗の詳細がメールで送信されます。チュートリアル全体を通じて、パラメーターを渡す方法が示されます。

次の図は、シナリオの概要を示しています。

コピー対象の Azure Blob Storage を示す図。成功した場合は詳細を含む電子メールを送信し、失敗した場合はエラーの詳細を含む電子メールを送信する。

このチュートリアルでは、以下のタスクの実行方法を説明します。

Data Factory の作成
Azure Storage のリンクされたサービスを作成する
Azure BLOB データセットを作成します。
コピーアクティビティと Web アクティビティを含むパイプラインを作成します。
アクティビティの出力を後続のアクティビティに送信します。
パラメーターの引き渡しとシステム変数を使用する
パイプラインの実行を開始します。
パイプラインとアクティビティの実行を監視します。

このチュートリアルでは .NET SDK を使用します。 Azure Data Factory の操作にはその他のメカニズムを使用することもできます。 Data Factory のクイックスタートについては、5 分間のクイックスタートを参照してください。

Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。

前提条件

Azure ストレージアカウント。 BLOB ストレージをソースデータストアとして使用します。 Azure ストレージアカウントがない場合は、「ストレージアカウントの作成」を参照してください。
Azure Storage Explorer. このツールをインストールするには、「Azure Storage Explorer」を参照してください。
Azure SQL Database。データベースをシンクデータストアとして使用します。 Azure SQL Database のデータベースがない場合は、Azure SQL Database のデータベースの作成に関するページを参照してください。
見ることができます。この記事では、Visual Studio 2019 を使用します。
Azure .NET SDK。 Azure .NET SDK をダウンロードしてインストールします。

現在 Data Factory が利用可能な Azure リージョンの一覧については、「リージョン別の利用可能な製品」を参照してください。データストアとコンピューティングは、別のリージョンに配置できます。ストアには、Azure Storage と Azure SQL Database が含まれます。コンピューティングには、Data Factory で使用される HDInsight が含まれます。

「Microsoft Entra アプリケーションを作成する」で説明されているようにアプリケーションを作成します。同じ記事の手順に従って、このアプリケーションを共同作成者ロールに割り当てます。このチュートリアルの後の方で、アプリケーション (クライアント) ID やディレクトリ (テナント) ID など、いくつかの値が必要になります。

BLOB テーブルを作成する

テキストエディターを開きます。次のテキストをコピーし、input.txt としてローカルに保存します。
```
Ethel|Berg
Tamika|Walsh
```
Azure ストレージエクスプローラーを開きます。お使いのストレージアカウントを展開します。 [BLOB コンテナー] を右クリックし、 [BLOB コンテナーの作成] を選択します。
新しいコンテナーに adfv2branch という名前を付け、 [アップロード] を選択して、そのコンテナーに input.txt ファイルを追加します。

Visual Studio プロジェクトを作成する

C# .NET コンソールアプリケーションを作成します。

Visual Studio を起動し、 [新しいプロジェクトの作成] を選択します。
[新しいプロジェクトの作成] で、C# の [コンソールアプリ (.NET Framework)] を選択してから、 [次へ] を選択します。
プロジェクトに ADFv2BranchTutorial という名前を付けます。
.NET バージョン 4.5.2 以降を選択し、 [作成] を選択します。

NuGet パッケージのインストール

[ツール]>[NuGet パッケージマネージャー]>[パッケージマネージャーコンソール] の順に選択します。
パッケージマネージャーコンソールで、次のコマンドを実行してパッケージをインストールします。詳細については、Microsoft.Azure.Management.DataFactory nuget パッケージを参照してください。
```
Install-Package Microsoft.Azure.Management.DataFactory
Install-Package Microsoft.Azure.Management.ResourceManager -IncludePrerelease
Install-Package Microsoft.IdentityModel.Clients.ActiveDirectory
```

データファクトリクライアントを作成する

Program.cs を開き、次のステートメントを追加します。

using System;
using System.Collections.Generic;
using System.Linq;
using Microsoft.Rest;
using Microsoft.Azure.Management.ResourceManager;
using Microsoft.Azure.Management.DataFactory;
using Microsoft.Azure.Management.DataFactory.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory;

Program クラスに次の静的変数を追加します。プレースホルダーは実際の値に置き換えてください。

// Set variables
static string tenantID = "<tenant ID>";
static string applicationId = "<application ID>";
static string authenticationKey = "<Authentication key for your application>";
static string subscriptionId = "<Azure subscription ID>";
static string resourceGroup = "<Azure resource group name>";

static string region = "East US";
static string dataFactoryName = "<Data factory name>";

// Specify the source Azure Blob information
static string storageAccount = "<Azure Storage account name>";
static string storageKey = "<Azure Storage account key>";
// confirm that you have the input.txt file placed in th input folder of the adfv2branch container.
static string inputBlobPath = "adfv2branch/input";
static string inputBlobName = "input.txt";
static string outputBlobPath = "adfv2branch/output";
static string emailReceiver = "<specify email address of the receiver>";

static string storageLinkedServiceName = "AzureStorageLinkedService";
static string blobSourceDatasetName = "SourceStorageDataset";
static string blobSinkDatasetName = "SinkStorageDataset";
static string pipelineName = "Adfv2TutorialBranchCopy";

static string copyBlobActivity = "CopyBlobtoBlob";
static string sendFailEmailActivity = "SendFailEmailActivity";
static string sendSuccessEmailActivity = "SendSuccessEmailActivity";

Main メソッドに次のコードを追加します。このコードによって、DataFactoryManagementClient クラスのインスタンスが作成されます。その後、このオブジェクトを使用して、データファクトリ、リンクされたサービス、データセット、パイプラインを作成します。また、このオブジェクトを使用して、パイプラインの実行の詳細を監視することもできます。

// Authenticate and create a data factory management client
var context = new AuthenticationContext("https://login.windows.net/" + tenantID);
ClientCredential cc = new ClientCredential(applicationId, authenticationKey);
AuthenticationResult result = context.AcquireTokenAsync("https://management.azure.com/", cc).Result;
ServiceClientCredentials cred = new TokenCredentials(result.AccessToken);
var client = new DataFactoryManagementClient(cred) { SubscriptionId = subscriptionId };

Data Factory の作成

Program.cs ファイルに CreateOrUpdateDataFactory メソッドを追加します。

static Factory CreateOrUpdateDataFactory(DataFactoryManagementClient client)
{
    Console.WriteLine("Creating data factory " + dataFactoryName + "...");
    Factory resource = new Factory
    {
        Location = region
    };
    Console.WriteLine(SafeJsonConvert.SerializeObject(resource, client.SerializationSettings));

    Factory response;
    {
        response = client.Factories.CreateOrUpdate(resourceGroup, dataFactoryName, resource);
    }

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

データファクトリを作成する次の行を Main メソッドに追加します。
```
Factory df = CreateOrUpdateDataFactory(client);
```

Azure Storage のリンクされたサービスを作成する

Program.cs ファイルに StorageLinkedServiceDefinition メソッドを追加します。

static LinkedServiceResource StorageLinkedServiceDefinition(DataFactoryManagementClient client)
{
   Console.WriteLine("Creating linked service " + storageLinkedServiceName + "...");
   AzureStorageLinkedService storageLinkedService = new AzureStorageLinkedService
   {
       ConnectionString = new SecureString("DefaultEndpointsProtocol=https;AccountName=" + storageAccount + ";AccountKey=" + storageKey)
   };
   Console.WriteLine(SafeJsonConvert.SerializeObject(storageLinkedService, client.SerializationSettings));
   LinkedServiceResource linkedService = new LinkedServiceResource(storageLinkedService, name:storageLinkedServiceName);
   return linkedService;
}

Azure Storage のリンクされたサービスを作成する次の行を Main メソッドに追加します。

client.LinkedServices.CreateOrUpdate(resourceGroup, dataFactoryName, storageLinkedServiceName, StorageLinkedServiceDefinition(client));

サポートされているプロパティと詳細については、「リンクされたサービスのプロパティ」を参照してください。

データセットを作成する

このセクションでは、ソース用とシンク用に 1 つずつ、2 つのデータセットを作成します。

ソース Azure BLOB 用のデータセットの作成

Azure BLOB データセットを作成するメソッドを追加します。サポートされているプロパティと詳細については、Azure BLOB データセットのプロパティに関するページを参照してください。

Program.cs ファイルに SourceBlobDatasetDefinition メソッドを追加します。

static DatasetResource SourceBlobDatasetDefinition(DataFactoryManagementClient client)
{
    Console.WriteLine("Creating dataset " + blobSourceDatasetName + "...");
    AzureBlobDataset blobDataset = new AzureBlobDataset
    {
        FolderPath = new Expression { Value = "@pipeline().parameters.sourceBlobContainer" },
        FileName = inputBlobName,
        LinkedServiceName = new LinkedServiceReference
        {
            ReferenceName = storageLinkedServiceName
        }
    };
    Console.WriteLine(SafeJsonConvert.SerializeObject(blobDataset, client.SerializationSettings));
    DatasetResource dataset = new DatasetResource(blobDataset, name:blobSourceDatasetName);
    return dataset;
}

Azure BLOB 内のソースデータを表すデータセットを定義します。この BLOB データセットは、前の手順でサポートされていた Azure Storage のリンクされたサービスを参照します。 BLOB データセットには、コピー元の BLOB の場所が記述されます。FolderPath と FileName が使用されます。

FolderPath ではパラメーターが使用されていることに注意してください。 sourceBlobContainer はパラメーターの名前であり、この式はパイプラインの実行で渡された値に置き換えられます。パラメーターを定義する構文は @pipeline().parameters.<parameterName> です。

シンク Azure BLOB 用のデータセットの作成

Program.cs ファイルに SourceBlobDatasetDefinition メソッドを追加します。

static DatasetResource SinkBlobDatasetDefinition(DataFactoryManagementClient client)
{
    Console.WriteLine("Creating dataset " + blobSinkDatasetName + "...");
    AzureBlobDataset blobDataset = new AzureBlobDataset
    {
        FolderPath = new Expression { Value = "@pipeline().parameters.sinkBlobContainer" },
        LinkedServiceName = new LinkedServiceReference
        {
            ReferenceName = storageLinkedServiceName
        }
    };
    Console.WriteLine(SafeJsonConvert.SerializeObject(blobDataset, client.SerializationSettings));
    DatasetResource dataset = new DatasetResource(blobDataset, name: blobSinkDatasetName);
    return dataset;
}

Azure BLOB のソースデータセットとシンクデータセット両方を作成する次のコードを Main メソッドに追加します。

client.Datasets.CreateOrUpdate(resourceGroup, dataFactoryName, blobSourceDatasetName, SourceBlobDatasetDefinition(client));

client.Datasets.CreateOrUpdate(resourceGroup, dataFactoryName, blobSinkDatasetName, SinkBlobDatasetDefinition(client));

C# クラスを作成する:EmailRequest

C# プロジェクトで、EmailRequest という名前のクラスを作成します。このクラスは、パイプラインがメールを送信するときに要求本文で送信するプロパティを定義します。このチュートリアルでは、パイプラインから電子メールに次の 4 つのプロパティが送信されます。

Message. メールの本文。コピーが成功した場合、このプロパティには、書き込まれたデータの量が含まれます。コピーが失敗した場合、このプロパティには、エラーの詳細が含まれます。
データファクトリ名。データファクトリの名前。
パイプライン名。パイプラインの名前。
受信者。パススルーされるパラメーター。このプロパティは、電子メールの受信者を指定します。

    class EmailRequest
    {
        [Newtonsoft.Json.JsonProperty(PropertyName = "message")]
        public string message;

        [Newtonsoft.Json.JsonProperty(PropertyName = "dataFactoryName")]
        public string dataFactoryName;

        [Newtonsoft.Json.JsonProperty(PropertyName = "pipelineName")]
        public string pipelineName;

        [Newtonsoft.Json.JsonProperty(PropertyName = "receiver")]
        public string receiver;

        public EmailRequest(string input, string df, string pipeline, string receiverName)
        {
            message = input;
            dataFactoryName = df;
            pipelineName = pipeline;
            receiver = receiverName;
        }
    }

電子メールワークフローエンドポイントを作成する

メール送信をトリガーするには、Azure Logic Apps を使用してワークフローを定義します。詳細については、「従量課金ロジックアプリワークフローの例を作成する」をご覧ください。

成功電子メールのワークフロー

Azure portal で、CopySuccessEmail という名前のロジックアプリワークフローを作成します。 [HTTP 要求の受信時] という名前の要求トリガーを追加します。要求トリガーで、[要求本文の JSON スキーマ] ボックスに、次の JSON を入力します。

{
    "properties": {
        "dataFactoryName": {
            "type": "string"
        },
        "message": {
            "type": "string"
        },
        "pipelineName": {
            "type": "string"
        },
        "receiver": {
            "type": "string"
        }
    },
    "type": "object"
}

ワークフローは次の例のようになります。

成功電子メールのワークフロー

この JSON コンテンツは、前のセクションで作成した EmailRequest クラスに合わせたものです。

[メールの送信] という名前の [Office 365 Outlook] アクションを追加します。このアクションの場合、[要求本文の JSON スキーマ] で渡されるプロパティを使用して、メールの書式設定をカスタマイズします。次に例を示します。

[メールの送信] という名前のアクションがあるワークフローデザイナー。

ワークフローを保存した後、トリガーの [HTTP POST の URL] の値をコピーして保存します。

失敗電子メールのワークフロー

CopySuccessEmail ロジックアプリワークフローを複製して、CopyFailEmail という名前の新しいワークフローにします。要求トリガーでは、[要求本文の JSON スキーマ] は同じです。 Subject のように電子メールの書式を変更し、失敗電子メールになるように調整します。たとえば次のようになります。

ワークフローデザイナーと失敗メールのワークフロー。

ワークフローを保存した後、トリガーの [HTTP POST の URL] の値をコピーして保存します。

これで、次の例のように、2 つのワークフローの URL を取得できたはずです。

//Success Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

//Fail Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

パイプラインを作成する

Visual Studio でプロジェクトに戻ります。ここで、コピーアクティビティと DependsOn プロパティを含むパイプラインを作成する次のコードを追加します。このチュートリアルでは、パイプラインに 1 つのアクティビティ (コピーアクティビティ) が含まれています。このアクティビティは、ソースとして BLOB データセットを取り込み、シンクとして別の BLOB データセットを取り込みます。このコピーアクティビティが成功したか失敗したかに応じて、異なるメールタスクが呼び出されます。

このパイプラインでは、次の機能を使用します。

パラメーター
Web アクティビティ
アクティビティの依存関係
あるアクティビティからの出力を別のアクティビティへの入力として使用する

このメソッドをプロジェクトに追加します。以降のセクションで、さらに詳しく説明します。

static PipelineResource PipelineDefinition(DataFactoryManagementClient client)
        {
            Console.WriteLine("Creating pipeline " + pipelineName + "...");
            PipelineResource resource = new PipelineResource
            {
                Parameters = new Dictionary<string, ParameterSpecification>
                {
                    { "sourceBlobContainer", new ParameterSpecification { Type = ParameterType.String } },
                    { "sinkBlobContainer", new ParameterSpecification { Type = ParameterType.String } },
                    { "receiver", new ParameterSpecification { Type = ParameterType.String } }

                },
                Activities = new List<Activity>
                {
                    new CopyActivity
                    {
                        Name = copyBlobActivity,
                        Inputs = new List<DatasetReference>
                        {
                            new DatasetReference
                            {
                                ReferenceName = blobSourceDatasetName
                            }
                        },
                        Outputs = new List<DatasetReference>
                        {
                            new DatasetReference
                            {
                                ReferenceName = blobSinkDatasetName
                            }
                        },
                        Source = new BlobSource { },
                        Sink = new BlobSink { }
                    },
                    new WebActivity
                    {
                        Name = sendSuccessEmailActivity,
                        Method = WebActivityMethod.POST,
                        Url = "https://prodxxx.eastus.logic.azure.com:443/workflows/00000000000000000000000000000000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=0000000000000000000000000000000000000000000000",
                        Body = new EmailRequest("@{activity('CopyBlobtoBlob').output.dataWritten}", "@{pipeline().DataFactory}", "@{pipeline().Pipeline}", "@pipeline().parameters.receiver"),
                        DependsOn = new List<ActivityDependency>
                        {
                            new ActivityDependency
                            {
                                Activity = copyBlobActivity,
                                DependencyConditions = new List<String> { "Succeeded" }
                            }
                        }
                    },
                    new WebActivity
                    {
                        Name = sendFailEmailActivity,
                        Method =WebActivityMethod.POST,
                        Url = "https://prodxxx.eastus.logic.azure.com:443/workflows/000000000000000000000000000000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=0000000000000000000000000000000000000000000",
                        Body = new EmailRequest("@{activity('CopyBlobtoBlob').error.message}", "@{pipeline().DataFactory}", "@{pipeline().Pipeline}", "@pipeline().parameters.receiver"),
                        DependsOn = new List<ActivityDependency>
                        {
                            new ActivityDependency
                            {
                                Activity = copyBlobActivity,
                                DependencyConditions = new List<String> { "Failed" }
                            }
                        }
                    }
                }
            };
            Console.WriteLine(SafeJsonConvert.SerializeObject(resource, client.SerializationSettings));
            return resource;
        }

パイプラインを作成する次の行を Main メソッドに追加します。

client.Pipelines.CreateOrUpdate(resourceGroup, dataFactoryName, pipelineName, PipelineDefinition(client));

パラメーター

パイプラインコードの最初のセクションでは、パラメーターを定義します。

sourceBlobContainer ソース BLOB データセットは、パイプラインでこのパラメーターを使用します。
sinkBlobContainer シンク BLOB データセットは、パイプラインでこのパラメーターを使用します。
receiver 成功または失敗のメールを受信者に送信する、パイプライン内の 2 つの Web アクティビティがこのパラメーターを使用します。

Parameters = new Dictionary<string, ParameterSpecification>
    {
        { "sourceBlobContainer", new ParameterSpecification { Type = ParameterType.String } },
        { "sinkBlobContainer", new ParameterSpecification { Type = ParameterType.String } },
        { "receiver", new ParameterSpecification { Type = ParameterType.String } }
    },

Web アクティビティ

Web アクティビティを使用すると、任意の REST エンドポイントを呼び出すことができます。このアクティビティの詳細については、「Azure Data Factory の Web アクティビティ」を参照してください。このパイプラインでは、Web アクティビティを使用して Logic Apps のメールワークフローを呼び出します。 CopySuccessEmail ワークフローを呼び出す Web アクティビティと CopyFailWorkFlow を呼び出す Web アクティビティの 2 つ作成します。

        new WebActivity
        {
            Name = sendCopyEmailActivity,
            Method = WebActivityMethod.POST,
            Url = "https://prodxxx.eastus.logic.azure.com:443/workflows/12345",
            Body = new EmailRequest("@{activity('CopyBlobtoBlob').output.dataWritten}", "@{pipeline().DataFactory}", "@{pipeline().Pipeline}", "@pipeline().parameters.receiver"),
            DependsOn = new List<ActivityDependency>
            {
                new ActivityDependency
                {
                    Activity = copyBlobActivity,
                    DependencyConditions = new List<String> { "Succeeded" }
                }
            }
        }

Url プロパティには、Logic Apps ワークフローの HTTP POST URL エンドポイントを貼り付けます。 Body プロパティで、EmailRequest クラスのインスタンスを渡します。この電子メール要求には次のプロパティが含まれます。

Message. @{activity('CopyBlobtoBlob').output.dataWritten の値を渡します。前のコピーアクティビティのプロパティにアクセスし、dataWritten の値を渡します。失敗の場合、@{activity('CopyBlobtoBlob').error.message の代わりにエラー出力を渡します。
データファクトリ名。 @{pipeline().DataFactory} の値を渡します。このシステム変数を使用すると、対応するデータファクトリ名にアクセスできます。システム変数の一覧については、システム変数に関するページを参照してください。
パイプライン名。 @{pipeline().Pipeline} の値を渡します。このシステム変数を使用すると、対応するパイプライン名にアクセスできます。
受信者。 "@pipeline().parameters.receiver" の値を渡します。パイプラインパラメーターにアクセスします。

このコードにより、前のコピーアクティビティに依存する、アクティビティの依存関係が新しく作成されます。

パイプラインの実行を作成する

パイプラインの実行をトリガーする次のコードを Main メソッドに追加します。

// Create a pipeline run
Console.WriteLine("Creating pipeline run...");
Dictionary<string, object> arguments = new Dictionary<string, object>
{
    { "sourceBlobContainer", inputBlobPath },
    { "sinkBlobContainer", outputBlobPath },
    { "receiver", emailReceiver }
};

CreateRunResponse runResponse = client.Pipelines.CreateRunWithHttpMessagesAsync(resourceGroup, dataFactoryName, pipelineName, arguments).Result.Body;
Console.WriteLine("Pipeline run ID: " + runResponse.RunId);

Main クラス

最終的な Main メソッドは次のようになります。

// Authenticate and create a data factory management client
var context = new AuthenticationContext("https://login.windows.net/" + tenantID);
ClientCredential cc = new ClientCredential(applicationId, authenticationKey);
AuthenticationResult result = context.AcquireTokenAsync("https://management.azure.com/", cc).Result;
ServiceClientCredentials cred = new TokenCredentials(result.AccessToken);
var client = new DataFactoryManagementClient(cred) { SubscriptionId = subscriptionId };

Factory df = CreateOrUpdateDataFactory(client);

client.LinkedServices.CreateOrUpdate(resourceGroup, dataFactoryName, storageLinkedServiceName, StorageLinkedServiceDefinition(client));
client.Datasets.CreateOrUpdate(resourceGroup, dataFactoryName, blobSourceDatasetName, SourceBlobDatasetDefinition(client));
client.Datasets.CreateOrUpdate(resourceGroup, dataFactoryName, blobSinkDatasetName, SinkBlobDatasetDefinition(client));

client.Pipelines.CreateOrUpdate(resourceGroup, dataFactoryName, pipelineName, PipelineDefinition(client));

Console.WriteLine("Creating pipeline run...");
Dictionary<string, object> arguments = new Dictionary<string, object>
{
    { "sourceBlobContainer", inputBlobPath },
    { "sinkBlobContainer", outputBlobPath },
    { "receiver", emailReceiver }
};

CreateRunResponse runResponse = client.Pipelines.CreateRunWithHttpMessagesAsync(resourceGroup, dataFactoryName, pipelineName, arguments).Result.Body;
Console.WriteLine("Pipeline run ID: " + runResponse.RunId);

プログラムをビルドして実行し、パイプラインの実行をトリガーしてください。

パイプラインの実行を監視する

次のコードを Main メソッドに追加します。

// 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")
        System.Threading.Thread.Sleep(15000);
    else
        break;
}

このコードは、データのコピーが完了するまで、実行の状態を継続的に確認します。

コピーアクティビティの実行の詳細 (たとえば、読み書きされたデータのサイズ) を取得する次のコードを Main メソッドに追加します。

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

List<ActivityRun> activityRuns = client.ActivityRuns.ListByPipelineRun(
resourceGroup, dataFactoryName, runResponse.RunId, DateTime.UtcNow.AddMinutes(-10), DateTime.UtcNow.AddMinutes(10)).ToList();

if (pipelineRun.Status == "Succeeded")
{
    Console.WriteLine(activityRuns.First().Output);
    //SaveToJson(SafeJsonConvert.SerializeObject(activityRuns.First().Output, client.SerializationSettings), "ActivityRunResult.json", folderForJsons);
}
else
    Console.WriteLine(activityRuns.First().Error);

Console.WriteLine("\nPress any key to exit...");
Console.ReadKey();

コードの実行

アプリケーションをビルドして起動し、パイプラインの実行を確認します。

このアプリケーションでは、データファクトリ、リンクされたサービス、データセット、パイプライン、およびパイプラインの実行の作成の進捗状況が表示されます。その後、パイプラインの実行状態が確認されます。コピーアクティビティの実行の詳細と、データの読み取り/書き込みのサイズが表示されるまで待ちます。次に、Azure Storage Explorer などのツールを使用して、変数で指定したように BLOB が inputBlobPath から outputBlobPath にコピーされたことを確認します。

出力は次のサンプルのようになります。

Creating data factory DFTutorialTest...
{
  "location": "East US"
}
Creating linked service AzureStorageLinkedService...
{
  "type": "AzureStorage",
  "typeProperties": {
    "connectionString": "DefaultEndpointsProtocol=https;AccountName=***;AccountKey=***"
  }
}
Creating dataset SourceStorageDataset...
{
  "type": "AzureBlob",
  "typeProperties": {
    "folderPath": {
      "type": "Expression",
      "value": "@pipeline().parameters.sourceBlobContainer"
    },
    "fileName": "input.txt"
  },
  "linkedServiceName": {
    "type": "LinkedServiceReference",
    "referenceName": "AzureStorageLinkedService"
  }
}
Creating dataset SinkStorageDataset...
{
  "type": "AzureBlob",
  "typeProperties": {
    "folderPath": {
      "type": "Expression",
      "value": "@pipeline().parameters.sinkBlobContainer"
    }
  },
  "linkedServiceName": {
    "type": "LinkedServiceReference",
    "referenceName": "AzureStorageLinkedService"
  }
}
Creating pipeline Adfv2TutorialBranchCopy...
{
  "properties": {
    "activities": [
      {
        "type": "Copy",
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "BlobSink"
          }
        },
        "inputs": [
          {
            "type": "DatasetReference",
            "referenceName": "SourceStorageDataset"
          }
        ],
        "outputs": [
          {
            "type": "DatasetReference",
            "referenceName": "SinkStorageDataset"
          }
        ],
        "name": "CopyBlobtoBlob"
      },
      {
        "type": "WebActivity",
        "typeProperties": {
          "method": "POST",
          "url": "https://xxxx.eastus.logic.azure.com:443/workflows/... ",
          "body": {
            "message": "@{activity('CopyBlobtoBlob').output.dataWritten}",
            "dataFactoryName": "@{pipeline().DataFactory}",
            "pipelineName": "@{pipeline().Pipeline}",
            "receiver": "@pipeline().parameters.receiver"
          }
        },
        "name": "SendSuccessEmailActivity",
        "dependsOn": [
          {
            "activity": "CopyBlobtoBlob",
            "dependencyConditions": [
              "Succeeded"
            ]
          }
        ]
      },
      {
        "type": "WebActivity",
        "typeProperties": {
          "method": "POST",
          "url": "https://xxx.eastus.logic.azure.com:443/workflows/... ",
          "body": {
            "message": "@{activity('CopyBlobtoBlob').error.message}",
            "dataFactoryName": "@{pipeline().DataFactory}",
            "pipelineName": "@{pipeline().Pipeline}",
            "receiver": "@pipeline().parameters.receiver"
          }
        },
        "name": "SendFailEmailActivity",
        "dependsOn": [
          {
            "activity": "CopyBlobtoBlob",
            "dependencyConditions": [
              "Failed"
            ]
          }
        ]
      }
    ],
    "parameters": {
      "sourceBlobContainer": {
        "type": "String"
      },
      "sinkBlobContainer": {
        "type": "String"
      },
      "receiver": {
        "type": "String"
      }
    }
  }
}
Creating pipeline run...
Pipeline run ID: 00000000-0000-0000-0000-0000000000000
Checking pipeline run status...
Status: InProgress
Status: InProgress
Status: Succeeded
Checking copy activity run details...
{
  "dataRead": 20,
  "dataWritten": 20,
  "copyDuration": 4,
  "throughput": 0.01,
  "errors": [],
  "effectiveIntegrationRuntime": "DefaultIntegrationRuntime (East US)"
}
{}

Press any key to exit...

このチュートリアルでは、次のタスクを実行しました。

Data Factory の作成
Azure Storage のリンクされたサービスを作成する
Azure BLOB データセットを作成します。
コピーアクティビティと Web アクティビティを含むパイプラインを作成します。
アクティビティの出力を後続のアクティビティに送信します。
パラメーターの引き渡しとシステム変数を使用する
パイプラインの実行を開始します。
パイプラインとアクティビティの実行を監視します。

これで、概念セクションに進むと、Azure Data Factory の詳細を確認できるようになりました。

パイプラインとアクティビティ

次の方法で共有

Data Factory パイプラインでのアクティビティの分岐と連鎖

前提条件

BLOB テーブルを作成する

Visual Studio プロジェクトを作成する

NuGet パッケージのインストール

データファクトリクライアントを作成する

Data Factory の作成

Azure Storage のリンクされたサービスを作成する

データセットを作成する

ソース Azure BLOB 用のデータセットの作成

シンク Azure BLOB 用のデータセットの作成

C# クラスを作成する:EmailRequest

電子メールワークフローエンドポイントを作成する

成功電子メールのワークフロー

失敗電子メールのワークフロー

パイプラインを作成する

パラメーター

Web アクティビティ

パイプラインの実行を作成する

Main クラス

パイプラインの実行を監視する

コードの実行

フィードバック

その他のリソース

次の方法で共有

Data Factory パイプラインでのアクティビティの分岐と連鎖

前提条件

BLOB テーブルを作成する

Visual Studio プロジェクトを作成する

NuGet パッケージのインストール

データ ファクトリ クライアントを作成する

Data Factory の作成

Azure Storage のリンクされたサービスを作成する

データセットを作成する

ソース Azure BLOB 用のデータセットの作成

シンク Azure BLOB 用のデータセットの作成

C# クラスを作成する:EmailRequest

電子メール ワークフロー エンドポイントを作成する

成功電子メールのワークフロー

失敗電子メールのワークフロー

パイプラインを作成する

パラメーター

Web アクティビティ

パイプラインの実行を作成する

Main クラス

パイプラインの実行を監視する

コードの実行

関連するコンテンツ

フィードバック

その他のリソース

データファクトリクライアントを作成する

電子メールワークフローエンドポイントを作成する