Azure Data Factoryのコピーアクティビティでスキーマ構造を動的に定義する

こんばんは。今日は、Azure Data Factoryのコピーアクティビティで、スキーマ構造を動的に定義する方法を試してみたので、備忘メモを残しておきたいと思います。

それではまいります。

やりたいこと

例えばAzure Blob Storage上に配置したCSVデータをAzure SQL Databaseのテーブルへコピーするシナリオで、CSVとテーブル間のマッピングを動的に指定できるようにしたい。

実装方法

今回は、STEPを2つに分けて実装を検討してみました。

STEP1では、まずは素直に要件を実現すること（＝スキーマを動的に指定できるようにすること）を考えます。

STEP2では、STEP1の実装をもう一歩、実際の運用を想定した実装に改良してみたいと思います。

STEP1：スキーマを動的に指定できるようにする

これは結論、コピーアクティビティのマップ設定をパラメータ化 & パイプライン実行時、そのパラメータにマッピングを定義したオブジェクトを指定するようにすればいけます。

公式ドキュメント上の解説はここ。

コピー アクティビティでのスキーマとデータ型のマッピング – Azure Data Factory & Azure Synapse | Microsoft Docs

以下実装方法。

まず、コピーアクティビティのマップタブから、「動的コンテンツの追加」に進む。

パイプライン式ビルダーで、新しいObject型のパラメータを追加する。

追加したパラメータを選択すると、自動的にパイプライン式が生成される。

これでパイプラインの設定は終わり。

この状態でパイプラインを実行（今回はデバッグを実行）すると、パラメータ値の指定を求められるので、今回マップに設定しているパラメータ（mapping4）に、スキーマ構造を渡す。

スキーマ構造の定義は、公式ドキュメントにサンプルがあるので、それを参考に組み立てる。

{
    "type": "TabularTranslator",
    "mappings": [
        {
            "source": {
                "name": "Id"
            },
            "sink": {
                "name": "CustomerID"
            }
        },
        {
            "source": {
                "name": "Name"
            },
            "sink": {
                "name": "LastName"
            }
        },
        {
            "source": {
                "name": "LastModifiedDate"
            },
            "sink": {
                "name": "ModifiedDate"
            }
        }
    ]
}

今回自分はヘッダのないCSVファイルをソースにしていたので、source側の列名設定の仕方が少し変わって以下のように。

{
  "type": "TabularTranslator",
  "mappings": [
    {
      "source": { "ordinal": 1 },
      "sink": {
        "name": "SinkColumnName1"
      }
    },
    {
      "source": { "ordinal": 2 },
      "sink": {
        "name": "SinkColumnName2"
      }
    },

　　　・・・
  ]
}

ソースのヘッダがない場合は、1,2のような序数で列を指定する必要があるらしい。

ヘッダー行なしの区切りテキスト ファイルからデータをコピーするには、列を名前ではなく序数で表します。

コピー アクティビティでのスキーマとデータ型のマッピング – Azure Data Factory & Azure Synapse | Microsoft Docs

これをパラメータとして渡すときは、以下のようにちょっと変わった書き方をする必要があるのですね。これは、一度パラメータを使わないマッピングを構成してみて、そのARMテンプレートをダウンロードして確認した。

"source": { "ordinal": 2 },

STEP2：パラメータを毎回自動的に付与できるようにする

さて、STEP1の実装で、スキーマを動的に指定できるようになりましたが、実運用では、パイプラインの実行都度、このパラメータを渡してやる必要があるかと思います。

渡す方法として一般的なのは、Data Factoryのパイプライン実行 APIを利用する方法でしょうか。

パイプラインの実行とトリガー – Azure Data Factory & Azure Synapse | Microsoft Docs

ただ、この場合APIをキックする別のしくみが必要になってきそうです。

何か、Data Factoryの中で完結できるような方法はないかなぁと思って探したら、使えそうなものがありました。

「Lookup Activity」です。

ルックアップ アクティビティ – Azure Data Factory & Azure Synapse | Microsoft Docs

これを使うと、任意のデータの内容を取得して、後続のアクティビティに値を渡すことができるようになります。

なので、スキーマ構造を定義したファイルを別途事前に用意しておき、コピーアクティビティの前にLookupアクティビティでそのファイル内容を読み取り、それを後続のコピーアクティビティのマップ定義のパラメータとして渡すようにすれば要件を実現できそうです。

実装方法は以下。

あらかじめ、先ほどのスキーマ構造を書きだしたファイルをAzure Blob Storage (or Data Lake Storage)などに置いておきます。

そのデータを参照するLookupアクティビティをコピーアクティビティの前に配置。

コピーアクティビティ側では、STEP1で動的に定義できるようにしたマップの定義式を以下のように変更します。これで、Lookupアクティビティの結果を使えるようになります。

ルックアップ アクティビティ – Azure Data Factory & Azure Synapse | Microsoft Docs

@{activity('Lookup1').output.firstRow

以上、Azure Data Factoryのコピーアクティビティでスキーマ構造を動的に定義する方法のメモでした。勉強になりました。

もし参考になった！という方は、下のいいねボタンをポチって頂けますと励みになります！ご覧いただきありがとうございました。

おしまい