データ ファクトリのアクティビティとパイプラインを作成する

  • 9 分

Azure Data Factory 内のアクティビティは、データに対して実行されるアクションを定義し、次の 3 つのカテゴリがあります。

  • データ移動アクティビティ
  • データ変換アクティビティ
  • アクティビティを制御する

データ移動アクティビティ

データ移動アクティビティは、データ ストア間でデータを移動するだけです。 コピー アクティビティを使用して、データ移動アクティビティを実行することも、JSON を使用して実行することもできます。 ソースおよびシンクとしてサポートされているさまざまなデータ ストアがあります。 この一覧は増え続け、 最新情報はこちらで確認できます。

データ変換アクティビティ

データ変換アクティビティは、マッピング データ フローを使用して、Azure Data Factory の作成ツール内でネイティブに実行できます。 または、コンピューティング リソースを呼び出して、変換によってデータを変更または拡張したり、データの分析を実行したりできます。 これには、Azure Databricks、Azure Batch、SQL Database、Azure Synapse Analytics、Machine Learning Services、Azure Virtual Machines、HDInsight などのコンピューティング テクノロジが含まれます。 カタログに格納されている既存の SQL Server Integration Services (SSIS) パッケージを使用して、Azure で実行できます

このリストは常に進化しており、 ここで最新情報を入手できます。

アクティビティを制御する

ADF ソリューションをグラフィカルに作成する場合は、設計内の制御フローを使用して、パイプライン アクティビティを調整できます。これには、シーケンスでのアクティビティのチェーン、分岐、パイプライン レベルでのパラメーターの定義、パイプラインのオンデマンド呼び出し中またはトリガーからの引数の渡しが含まれます。 現在の機能は次のとおりです。

管理活動 説明
Execute Pipeline アクティビティ パイプラインの実行アクティビティを使用すると、Data Factory パイプラインで別のパイプラインを呼び出すことができます。
ForEachActivity ForEach アクティビティは、パイプライン内に反復的な制御フローを定義します。 このアクティビティは、コレクションを反復処理し、ループ内で指定されたアクティビティを実行するために使用されます。 このアクティビティのループ実装は、プログラミング言語の Foreach ループ構造に似ています。
WebActivity Web アクティビティを使用して、Data Factory パイプラインからカスタム REST エンドポイントを呼び出すことができます。 このアクティビティにおける利用やアクセスのために、データセットやリンクされたサービスを渡すことができます。
ルックアップ アクティビティ Lookup アクティビティは、任意の外部ソースから得られるレコード/テーブル名/値の読み取りや検索に使用できます。 この出力は、以後のアクティビティによって後から参照されることがあります。
メタデータの取得アクティビティ GetMetadata アクティビティを使用して、Azure Data Factory 内の任意のデータのメタデータを取得できます。
Until アクティビティ プログラミング言語の Do-Until ループ構造に似た、Do-Until ループを実装します。 アクティビティに関連付けられている条件が true に評価されるまで、ループ内で一連のアクティビティを実行します。 Data Factory の until アクティビティのタイムアウト値を指定できます。
If Condition アクティビティ If Condition は、条件の評価結果が true か false かに基づいて制御を分岐させるために使用できます。 If Condition アクティビティには、プログラミング言語における if ステートメントと同じ働きがあります。 条件が true に評価されたときにアクティビティのセットを評価し、条件が false に評価されたときに別のアクティビティのセットを評価します。
Wait アクティビティ パイプラインで Wait アクティビティを使用すると、パイプラインは指定された期間待機してから、後続のアクティビティの実行を続行します。

最新情報はこちらで入手できます。

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

アクティビティの定義

JSON 表記を使用する場合、activities セクションには 1 つ以上のアクティビティを定義できます。 アクティビティには、主に実行アクティビティと制御アクティビティの 2 種類があります。 実行 (コンピューティングとも呼ばれます) アクティビティには、データ移動アクティビティとデータ変換アクティビティが含まれます。 これらのアクティビティには、次のような最上位構造があります。

{
    "name": "Execution Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "linkedServiceName": "MyLinkedService",
    "policy":
    {
    },
    "dependsOn":
    {
    }
}

以下の表に、JSON の上記のプロパティに関する説明を示します。

プロパティ 説明 必須
名前 アクティビティの名前。 イエス
説明 アクティビティまたは使用対象を説明するテキスト。 いいえ
型 アクティビティの種類を定義します。 イエス
リンクされたサービス名 アクティビティで使用される、リンクされたサービスの名前。 はい (HDInsight、Machine Learning バッチ スコアリング アクティビティ、ストアド プロシージャ アクティビティ)
typeProperties typeProperties セクションのプロパティは、アクティビティの種類に応じて異なります。 いいえ
ポリシー アクティビティの実行時の動作に影響するポリシーです。 このプロパティには、タイムアウトと再試行の動作が含まれます。 いいえ
dependsOn このプロパティを使用して、アクティビティの依存関係と、後続のアクティビティが前のアクティビティにどのように依存するかを定義します。 いいえ

制御アクティビティの定義

Data Factory のコントロール アクティビティは、次のように JSON 形式で定義されます。

{
    "name": "Control Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "dependsOn":
    {
    }
}

以下の表に、JSON の上記のプロパティに関する説明を示します。

プロパティ 説明 必須
名前 アクティビティの名前。 イエス
説明 アクティビティまたは使用対象を説明するテキスト。 イエス
型 アクティビティの種類を定義します。 イエス
typeProperties typeProperties セクションのプロパティは、アクティビティの種類に応じて異なります。 いいえ
dependsOn このプロパティを使用して、アクティビティの依存関係と、後続のアクティビティが前のアクティビティにどのように依存するかを定義します。 いいえ

パイプラインの定義

JSON 形式内にパイプラインを定義する方法を以下に示します。

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities":
        [
        ],
        "parameters": {
         }
    }
}

以下の表に、JSON の上記のプロパティに関する説明を示します。

プロパティ 説明 必須
名前 パイプラインの名前。 イエス
説明 パイプラインの用途を説明するテキスト。 いいえ
アクティビティ アクティビティ セクションには、1 つ以上のアクティビティを定義できます。 イエス
パラメーター parameters セクションには、パイプライン内で 1 つ以上のパラメーターを定義できるため、パイプラインを柔軟に再利用できます。 いいえ

例

次の JSON は、"MyFirstPipeline" という名前のパイプラインを定義します。このパイプラインには、"StorageLinkedService" という名前のリンクされたサービスに格納されているスクリプト名 "partitionweblogs.hql" からクエリを呼び出す HDInsightHive の 1 つのアクティビティの種類が含まれており、"AzureBlobInput" という名前の入力と "AzureBlobOutput" という名前の出力が含まれます。 これは、"HDInsightOnDemandLinkedService" という名前のリンクされたサービスで定義されているコンピューティング リソースに対して実行されます。

パイプラインは月単位で実行されるようにスケジュールされており、失敗した場合は 3 回実行を試みます。

{
    "name": "MyFirstPipeline",
    "properties": {
        "description": "My first Azure Data Factory pipeline",
        "activities": [
            {
                "type": "HDInsightHive",
                "typeProperties": {
                    "scriptPath": "adfgetstarted/script/partitionweblogs.hql",
                    "scriptLinkedService": "StorageLinkedService",
                    "defines": {
                        "inputtable": "wasb://adfgetstarted@ctostorageaccount.blob.core.windows.net/inputdata",
                        "partitionedtable": "wasb://adfgetstarted@ctostorageaccount.blob.core.windows.net/partitioneddata"
                    }
                },
                "inputs": [
                    {
                        "name": "AzureBlobInput"
                    }
                ],
                "outputs": [
                    {
                        "name": "AzureBlobOutput"
                    }
                ],
                "policy": {
                    "concurrency": 1,
                    "retry": 3
                },
                "scheduler": {
                    "frequency": "Month",
                    "interval": 1
              },
                "name": "RunSampleHiveActivity",
                "linkedServiceName": "HDInsightOnDemandLinkedService"
            }
        ],
        "start": "2017-04-01T00:00:00Z",
        "end": "2017-04-02T00:00:00Z",
        "isPaused": false,
        "hubName": "ctogetstarteddf_hub",
        "pipelineMode": "Scheduled"
    }
}