Azure Data Factory および Azure Synapse Analytics 内の Data Flow アクティビティ

[アーティクル]
10/20/2023

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

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

データフローアクティビティを使用して、Mapping Data Flow を介してデータを変換および移動します。データフローを初めて扱う場合は、Mapping Data Flow の概要に関するページを参照してください。

UI を使用して Data Flow アクティビティを作成する

パイプライン内で Data Flow アクティビティを使用するには、次の手順を実行します。

パイプラインの [アクティビティ] ペイン内で Data Flow を検索し、Data Flow アクティビティをパイプラインキャンバスにドラッグします。
キャンバス上で新しいデータフローアクティビティ (まだ選択されていない場合)、[設定] タブの順に選択して、詳細を編集します。
チェックポイントキーは、データフローが変更されたデータキャプチャに使用されるときにチェックポイントを設定するために使用されます。上書きすることができます。データフローアクティビティでは、"パイプライン名 + アクティビティ名" ではなくチェックポイントキーとして guid 値を使用するため、名前変更アクションがある場合でも、顧客の変更データキャプチャの状態を常に追跡できます。既存のすべてのデータフローアクティビティは、下位互換性のために古いパターンキーを使用します。変更データキャプチャが有効なデータフローリソースを使用して、新しいデータフローアクティビティをパブリッシュした後のチェックポイントキーのオプションを次に示します。
既存のデータフローを選択するか、[新規作成] ボタンを使用して新しく作成します。必要に応じてその他のオプションを選択し、構成を完了します。

構文

{
    "name": "MyDataFlowActivity",
    "type": "ExecuteDataFlow",
    "typeProperties": {
      "dataflow": {
         "referenceName": "MyDataFlow",
         "type": "DataFlowReference"
      },
      "compute": {
         "coreCount": 8,
         "computeType": "General"
      },
      "traceLevel": "Fine",
      "runConcurrently": true,
      "continueOnError": true,      
      "staging": {
          "linkedService": {
              "referenceName": "MyStagingLinkedService",
              "type": "LinkedServiceReference"
          },
          "folderPath": "my-container/my-folder"
      },
      "integrationRuntime": {
          "referenceName": "MyDataFlowIntegrationRuntime",
          "type": "IntegrationRuntimeReference"
      }
}

型のプロパティ

プロパティ	説明	使用できる値	必須
dataflow	実行されているデータフローへの参照	DataFlowReference	はい
integrationRuntime	データフローが実行されているコンピューティング環境です。指定されていない場合は、自動解決 Azure 統合ランタイムが使用されます。	IntegrationRuntimeReference	いいえ
compute.coreCount	Spark クラスター内で使用されるコアの数です。自動解決 Azure 統合ランタイムが使用されている場合にのみ指定できます	8、16、32、48、80、144、272	いいえ
compute.computeType	Spark クラスター内で使用されるコンピューティングの種類です。自動解決 Azure 統合ランタイムが使用されている場合にのみ指定できます	"一般"	いいえ
staging.linkedService	Azure Synapse Analytics ソースまたはシンクを使用している場合は、PolyBase ステージングに使用するストレージアカウントを指定します。 Azure Storage が VNet サービスエンドポイントを使用して構成されている場合は、ストレージアカウントで [信頼された Microsoft サービスを許可する] を有効にしたマネージド ID 認証を使用する必要があります。「Azure Storage で VNet サービスエンドポイントを使用した場合の影響」を参照してください。また、Azure Blob と Azure Data Lake Storage Gen2 に必要な構成についても説明します。	LinkedServiceReference	データフローが Azure Synapse Analytics に対して読み取りまたは書き込みを行う場合のみ
staging.folderPath	Azure Synapse Analytics ソースまたはシンクを使用している場合は、PolyBase ステージングに使用する BLOB ストレージアカウント内のフォルダーパス	String	データフローが Azure Synapse Analytics に対して読み取りまたは書き込みを行う場合のみ
traceLevel	データフローアクティビティの実行のログレベルを設定します	Fine、Coarse、None	No

データフローの実行

実行時、データフローコンピューティングのサイズを動的に設定する

Core Count プロパティと Compute Type プロパティは、実行時に入ってくるソースデータのサイズに合わせて調整されるよう、動的に設定できます。ソースデータセットデータのサイズを見つける目的で、Lookup や Get Metadata など、パイプラインアクティビティを使用します。次に、Data Flow アクティビティプロパティで Add Dynamic Content を使用します。小、中、または大規模のコンピューティングサイズを選択できます。必要に応じて、[カスタム] を選択し、コンピューティングの種類とコア数を手動で構成します。

動的データフロー

こちらの短い動画チュートリアルでこの手法について説明しています

データフロー統合ランタイム

データフローアクティビティの実行に使用する統合ランタイムを選択します。既定では、このサービスは 4 つのワーカーコアを持つ自動解決 Azure 統合ランタイムを使用します。この IR は汎用目的のコンピューティングの種類で、ご使用のサービスインスタンスと同じリージョンで実行します。運用可能なパイプラインとするには、データフローアクティビティの実行用に特定のリージョン、コンピューティングの種類、コア数、および TTL を定義する独自の Azure 統合ランタイムを作成することを強くお勧めします。

General Purpose の最小コンピューティングタイプが 8 + 8 (合計 16 個の v コア) の構成で、ほとんどの運用ワークロードの最小推奨は 10 分の Time to live (TTL) です。小さい TTL を設定することにより、Azure IR は、コールドクラスターのような数分の開始時間を要さないウォームクラスターを維持できます。詳細については、Azure 統合ランタイムに関するページを参照してください。

Azure Integration Runtime

重要

データフローアクティビティでの Integration Runtime の選択は、お使いのパイプラインのトリガー済みの実行のみに適用されます。データフローを使用したパイプラインのデバッグは、デバッグセッションで指定されたクラスターで実行されます。

PolyBase

Azure Synapse Analytics をシンクまたはソースとして使用する場合は、PolyBase バッチ読み込み用のステージングの場所を選択する必要があります。 PolyBase を使用すると、データを行ごとに読み込む代わりに一括してバッチ読み込みを行うことができます。 PolyBase を実行すると、Azure Synapse Analytics への読み込み時間が大幅に短縮されます。

チェックポイントキー

データフローソースの変更キャプチャオプションを使用すると、ADF はユーザーの代わりに自動的にチェックポイントの保持と管理を行います。既定のチェックポイントキーは、データフロー名とパイプライン名のハッシュです。ソーステーブルまたはフォルダーに動的パターンを使用する場合は、このハッシュを上書きして、独自のチェックポイントキーの値をここに設定することをお勧めします。

ログ記録レベル

データフローアクティビティのすべてのパイプライン実行がすべての詳細なテレメトリログを完全にログ記録する必要がない場合は、必要に応じてログレベルを "Basic" または "None" に設定できます。データフローを "Verbose" モード (既定値) で実行している場合、データ変換中に個別のパーティションレベルのそれぞれでアクティビティを完全にログ記録するように、サービスに要求していることになります。これは負荷の高い操作であるため、トラブルシューティングを行うときにのみ詳細を有効にすることで、データフローとパイプラインのパフォーマンス全体を向上させることができます。 "Basic" モードは変換の実行時間だけをログ記録し、"None" は実行時間の要約だけを記録します。

ログ記録レベル

シンクのプロパティ

データフローのグループ化機能を使用すると、シンクの実行順序を設定できるだけでなく、同じグループ番号を使用してシンクをグループ化できます。グループを管理しやすくするため、シンクを同じグループ内で並列で実行するように、サービスに要求できます。また、いずれかのシンクでエラーが発生しても続行するようにシンクグループを設定することもできます。

データフローシンクの既定の動作では、各シンクが逐次実行され、シンクでエラーが発生した場合はデータフローが失敗します。さらに、データフロープロパティでシンクに異なる優先順位を設定しない限り、すべてのシンクは既定で同じグループに設定されます。

シンクのプロパティ

First row only (先頭行のみ)

このオプションは、"アクティビティへの出力" でキャッシュシンクが有効になっているデータフローでのみ使用できます。パイプラインに直接挿入されるデータフローからの出力は、2 MB に制限されます。 "最初の行のみ" を設定すると、データフローアクティビティの出力をパイプラインに直接挿入する際に、データフローからのデータ出力を制限することができます。

データフローをパラメーター化する

パラメーター化されたデータセット

データフローでパラメーター化されたデータセットを使用する場合は、 [設定] タブでパラメーター値を設定します。

データフローの実行パラメーター

パラメーター化されたデータフロー

データフローがパラメーター化されている場合は、 [パラメーター] タブでデータフローパラメーターの動的な値を設定します。パイプライン式言語またはデータフロー式言語のいずれかを使用して、動的パラメーター値またはリテラルパラメーター値を割り当てることができます。詳しくは、データフローパラメーターに関するページを参照してください。

パラメーター化されたコンピューティングのプロパティ

自動解決 Azure 統合ランタイムを使用し、かつ compute.coreCount と compute.computeType の値を指定すると、コアカウントやコンピューティングの種類をパラメーター化することができます。

データフローの実行パラメーターの例

データフローアクティビティのパイプラインデバッグ

データフローアクティビティを使用してデバッグパイプラインを実行するには、上部バーにある [Data Flow Debug](データフローデバッグ) スライダーを使用して、データフローデバッグモードをオンに切り替える必要があります。デバッグモードでは、アクティブな Spark クラスターに対してデータフローを実行できます。詳細については、デバッグモードに関するページを参照してください。

デバッグパイプラインは、データフローアクティビティ設定で指定された統合ランタイム環境ではなく、アクティブなデバッグクラスターに対して実行されます。デバッグモードを開始するときに、デバッグコンピューティング環境を選択できます。

データフローアクティビティを監視する

データフローアクティビティには、パーティション分割、ステージ時間、およびデータ系列の情報を表示できる特別な監視エクスペリエンスがあります。 [アクション] の下にある眼鏡アイコンを使用して、[監視] ウィンドウを開きます。詳しくは、データフローの監視に関するページを参照してください。

後続のアクティビティでデータフローアクティビティの結果を使用する

データフローアクティビティは、各シンクに書き込まれた行の数と各ソースから読み取られた行に関するメトリックを出力します。これらの結果は、アクティビティの実行結果の output セクションに返されます。返されるメトリックは、以下の JSON の形式です。

{
    "runStatus": {
        "metrics": {
            "<your sink name1>": {
                "rowsWritten": <number of rows written>,
                "sinkProcessingTime": <sink processing time in ms>,
                "sources": {
                    "<your source name1>": {
                        "rowsRead": <number of rows read>
                    },
                    "<your source name2>": {
                        "rowsRead": <number of rows read>
                    },
                    ...
                }
            },
            "<your sink name2>": {
                ...
            },
            ...
        }
    }
}

たとえば、'dataflowActivity' という名前のアクティビティで、'sink1' という名前のシンクに書き込まれた行の数を取得するには、@activity('dataflowActivity').output.runStatus.metrics.sink1.rowsWritten を使用します。

このシンクで使用されていた、'source1' という名前のソースから読み取られた行の数を取得するには、@activity('dataflowActivity').output.runStatus.metrics.sink1.sources.source1.rowsRead を使用します。

Note

シンクに書き込まれた行が 0 の場合は、メトリックに表示されません。存在を確認するには、contains 関数を使用します。たとえば、contains(activity('dataflowActivity').output.runStatus.metrics, 'sink1') は、sink1 に何らかの行が書き込まれたかどうかを確認します。

サポートされている制御フローアクティビティを参照してください。

次の方法で共有

Azure Data Factory および Azure Synapse Analytics 内の Data Flow アクティビティ

UI を使用して Data Flow アクティビティを作成する

構文