Azure Data Factory または Azure Synapse Analytics を使用して Azure Databricks Delta Lake との間でデータをコピーします

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

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

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

この記事では、Azure Data Factory および Azure Synapse のコピーアクティビティを使用して、Azure Databricks Delta Lake をコピー先またはコピー元としてデータをコピーする方法について説明します。この記事は、コピーアクティビティの概要を示しているコピーアクティビティの記事に基づいています。

サポートされる機能

この Azure Databricks Delta Lake コネクタは、次の機能でサポートされます。

サポートされる機能	IR
Copy アクティビティ (ソース/シンク)	① ②
マッピングデータフロー (ソース/シンク)	①
Lookup アクティビティ	① ②

① Azure 統合ランタイム ② セルフホステッド統合ランタイム

一般に、このサービスは、さまざまなニーズを満たすために、次の機能を備えた Delta Lake をサポートしています。

コピーアクティビティでは、サポートされている任意のソースデータストアから Azure Databricks Delta Lake テーブル、および Delta Lake テーブルからサポートされている任意のシンクデータストアにデータをコピーするために、Azure Databricks Delta Lake コネクタがサポートされています。 Databricks クラスターを活用してデータ移動を実行します。詳細については、「前提条件」セクションを参照してください。
マッピングデータフローでは、ソースおよびシンクとして Azure Storage の汎用差分形式がサポートされています。これにより、コーディング不要の ETL の差分ファイルの読み取りと書き込みが可能になり、マネージド Azure Integration Runtime で実行されます。
Databricks アクティビティでは、Delta Lake 上でコード中心の ETL または機械学習ワークロードを調整することがサポートされています。

前提条件

この Azure Databricks Delta Lake コネクタを使用するには、Azure Databricks でクラスターを設定する必要があります。

Delta Lake にデータをコピーする場合、コピーアクティビティは Azure Databricks クラスターを呼び出して、Azure Storage からデータを読み取ります。これは元のソースまたはステージング領域で、サービスが組み込みのステージングコピーを介してソースデータを最初に書き込みます。詳細については、「シンクとしての Delta Lake」を参照してください。
同様に、Delta Lake からデータをコピーする場合、コピーアクティビティは Azure Databricks クラスターを呼び出して、Azure Storage にデータを書き込みます。これは元のシンクまたはステージング領域で、サービスが引き続き組み込みのステージングコピーを介して最終的なシンクにデータを書き込みます。詳細については、「ソースとしての Delta Lake」を参照してください。

Databricks クラスターは、Azure Blob または Azure Data Lake Storage Gen2 アカウントにアクセスできる必要があります。これは、ソース/シンク/ステージングに使用されるストレージコンテナー/ファイルシステムと、Data Lake テーブルを書き込むコンテナー/ファイルシステムの両方です。

Azure Data Lake Storage Gen2 を使用するために、Apache Spark 構成の一部として、Databricks クラスターでサービスプリンシパルを構成できます。サービスプリンシパルを使用した直接アクセスに関する記事の手順に従います。
Azure Blob Storage を使用するには、Apache Spark 構成の一部として、Databricks クラスターでストレージアカウントアクセスキーまたは SAS トークンを構成します。「RDD API を使用した Azure Blob Storage へのアクセス」の手順に従います。

コピーアクティビティの実行中、構成したクラスターが終了した場合は、自動的に開始されます。オーサリング UI を使用してパイプラインを作成する場合、データのプレビューなどの操作にはライブクラスターが必要ですが、クラスターが起動されることはありません。

クラスター構成の指定

[クラスターモード] ドロップダウンで、 [標準] を選択します。
[Databricks Runtime のバージョン] ドロップダウンで、Databricks Runtime のバージョンを選択します。

Spark 構成に次のプロパティを追加して、自動最適化をオンにします。

spark.databricks.delta.optimizeWrite.enabled true
spark.databricks.delta.autoCompact.enabled true

統合とスケーリングのニーズに応じて、クラスターを構成します。

クラスター構成の詳細については、「クラスターの構成」を参照してください。

はじめに

パイプラインでコピーアクティビティを実行するには、次のいずれかのツールまたは SDK を使用します。

UI を使って Azure Databricks Delta Lake へのリンクサービスを作成する

次の手順を使用して、Azure portal UI で Azure Databricks Delta Lake へのリンクサービスを作成します。

Azure Data Factory または Synapse ワークスペースの [管理] タブに移動し、[リンクサービス] を選択して、[新規] をクリックします。
- Azure Data Factory
- Azure Synapse
デルタを検索し、Azure Databricks Delta Lake コネクタを選択します。
サービスの詳細を構成し、接続をテストして、新しいリンクサービスを作成します。

コネクタの構成の詳細

次のセクションでは、Azure Databricks Delta Lake コネクタに固有のエンティティを定義するプロパティについて詳しく説明します。

リンクされたサービスのプロパティ

この Azure Databricks Delta Lake コネクタは、次の種類の認証をサポートしています。詳細については、対応するセクションをご覧ください。

アクセストークン
システム割り当てマネージド ID 認証
ユーザー割り当てマネージド ID 認証

アクセストークン

Azure Databricks Delta Lake のリンクされたサービスでは、次のプロパティがサポートされています。

プロパティ	内容	必須
type	type プロパティは、AzureDatabricksDeltaLake に設定する必要があります。	はい
domain	Azure Databricks ワークスペースの URL を指定します (例: `https://adb-xxxxxxxxx.xx.azuredatabricks.net`)。
clusterId	既存のクラスターのクラスター ID を指定します。これは作成済みの対話型クラスターでなければなりません。対話型クラスターのクラスター ID は Databricks ワークスペース -> クラスター -> 対話型クラスター名 -> 構成 -> タグで見つけることができます。詳細については、こちらを参照してください。
accessToken	サービスの Azure Databricks の認証にはアクセストークンが必要です。アクセストークンは、Databricks ワークスペースから生成する必要があります。アクセストークンを見つける詳細な手順については、こちらを参照してください。
connectVia	データストアに接続するために使用される統合ランタイム。 Azure 統合ランタイムまたはセルフホステッド統合ランタイムを使用できます (データストアがプライベートネットワークにある場合)。指定されていない場合は、既定の Azure 統合ランタイムが使用されます。	いいえ

例:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "accessToken": {
                "type": "SecureString", 
                "value": "<access token>"
          	}
        }
    }
}

システム割り当てマネージド ID 認証

Azure リソースのシステム割り当てマネージド ID について詳しくは、Azure リソースのシステム割り当てマネージド ID に関する記事を参照してください。

システム割り当てマネージド ID 認証を使用するには、次の手順に従ってアクセス許可を付与します。

データファクトリまたは Synapse ワークスペースと共に生成されたマネージド ID オブジェクト ID の値をコピーして、マネージド ID 情報を取得します。
Azure Databricks でマネージド ID に適切なアクセス許可を付与します。一般に、Azure Databricks のアクセス制御 (IAM) で、少なくとも共同作成者ロールをシステム割り当てマネージド ID に付与する必要があります。

Azure Databricks Delta Lake のリンクされたサービスでは、次のプロパティがサポートされています。

プロパティ	内容	必須
type	type プロパティは、AzureDatabricksDeltaLake に設定する必要があります。	はい
domain	Azure Databricks ワークスペースの URL を指定します (例: `https://adb-xxxxxxxxx.xx.azuredatabricks.net`)。	はい
clusterId	既存のクラスターのクラスター ID を指定します。これは作成済みの対話型クラスターでなければなりません。対話型クラスターのクラスター ID は Databricks ワークスペース -> クラスター -> 対話型クラスター名 -> 構成 -> タグで見つけることができます。詳細については、こちらを参照してください。	はい
workspaceResourceId	Azure Databricks のワークスペースリソース ID を指定します。	はい
connectVia	データストアに接続するために使用される統合ランタイム。 Azure 統合ランタイムまたはセルフホステッド統合ランタイムを使用できます (データストアがプライベートネットワークにある場合)。指定されていない場合は、既定の Azure 統合ランタイムが使用されます。	いいえ

例:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "workspaceResourceId": "<workspace resource id>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

ユーザー割り当てマネージド ID 認証

Azure リソースのユーザー割り当てマネージド ID について詳しくは、ユーザー割り当てマネージド ID に関する記事を参照してください。

ユーザー割り当てマネージド ID 認証を使用するには、次の手順に従います。

1 つ以上のユーザー割り当てマネージド ID を作成して、Azure Databricks でアクセス許可を付与します。一般に、Azure Databricks のアクセス制御 (IAM) で、少なくとも共同作成者ロールをユーザー割り当てマネージド ID に付与する必要があります。
1 つ以上のユーザー割り当てマネージド ID をデータファクトリまたは Synapse ワークスペースに割り当て、ユーザー割り当てマネージド ID ごとに資格情報を作成します。

Azure Databricks Delta Lake のリンクされたサービスでは、次のプロパティがサポートされています。

プロパティ	内容	必須
type	type プロパティは、AzureDatabricksDeltaLake に設定する必要があります。	はい
domain	Azure Databricks ワークスペースの URL を指定します (例: `https://adb-xxxxxxxxx.xx.azuredatabricks.net`)。	はい
clusterId	既存のクラスターのクラスター ID を指定します。これは作成済みの対話型クラスターでなければなりません。対話型クラスターのクラスター ID は Databricks ワークスペース -> クラスター -> 対話型クラスター名 -> 構成 -> タグで見つけることができます。詳細については、こちらを参照してください。	はい
資格情報	ユーザー割り当てマネージド ID を資格情報オブジェクトとして指定します。	はい
workspaceResourceId	Azure Databricks のワークスペースリソース ID を指定します。	はい
connectVia	データストアに接続するために使用される統合ランタイム。 Azure 統合ランタイムまたはセルフホステッド統合ランタイムを使用できます (データストアがプライベートネットワークにある場合)。指定されていない場合は、既定の Azure 統合ランタイムが使用されます。	いいえ

例:

{
    "name": "AzureDatabricksDeltaLakeLinkedService",
    "properties": {
        "type": "AzureDatabricksDeltaLake",
        "typeProperties": {
            "domain": "https://adb-xxxxxxxxx.xx.azuredatabricks.net",
            "clusterId": "<cluster id>",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            },
            "workspaceResourceId": "<workspace resource id>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

データセットのプロパティ

データセットを定義するために使用できるセクションとプロパティの完全な一覧については、データセットに関する記事をご覧ください。

Azure Databricks Delta Lake データセットでは、次のプロパティがサポートされます。

プロパティ	内容	必須
type	データセットの type プロパティには、AzureDatabricksDeltaLakeDataset を設定する必要があります。	はい
database	データベースの名前です。	ソースの場合はいいえ、シンクの場合ははい
table	デルタテーブルの名前。	ソースの場合はいいえ、シンクの場合ははい

例:

{
    "name": "AzureDatabricksDeltaLakeDataset",
    "properties": {
        "type": "AzureDatabricksDeltaLakeDataset",
        "typeProperties": {
            "database": "<database name>",
            "table": "<delta table name>"
        },
        "schema": [ < physical schema, optional, retrievable during authoring > ],
        "linkedServiceName": {
            "referenceName": "<name of linked service>",
            "type": "LinkedServiceReference"
        }
    }
}

コピーアクティビティのプロパティ

アクティビティの定義に利用できるセクションとプロパティの完全な一覧については、パイプラインに関する記事を参照してください。このセクションでは、Azure Databricks Delta Lake のソースとシンクでサポートされるプロパティの一覧を示します。

ソースとしての Delta Lake

Azure Databricks Delta Lake からデータをコピーするために、コピーアクティビティの source セクションでは次のプロパティがサポートされています。

プロパティ	内容	必須
type	コピーアクティビティのソースの type プロパティは AzureDatabricksDeltaLakeSource を設定する必要があります。	はい
query	データを読み取るための SQL クエリを指定します。タイムトラベル制御については、次のパターンに従います。 - `SELECT * FROM events TIMESTAMP AS OF timestamp_expression` - `SELECT * FROM events VERSION AS OF version`	いいえ
exportSettings	デルタテーブルからデータを取得するために使用される詳細設定。	いいえ
`exportSettings` の下:
type	エクスポートコマンドの種類。AzureDatabricksDeltaLakeExportCommand に設定します。	Yes
dateFormat	日付型を日付形式の文字列に書式設定します。カスタム日付形式は datetime パターンの形式に従います。指定しない場合は、既定値の `yyyy-MM-dd` が使用されます。	いいえ
timestampFormat	タイムスタンプ型をタイムスタンプ形式の文字列に書式設定します。カスタム日付形式は datetime パターンの形式に従います。指定しない場合は、既定値の `yyyy-MM-dd'T'HH:mm:ss[.SSS][XXX]` が使用されます。	いいえ

Delta Lake から直接コピーする

シンクのデータストアと形式がこのセクションで説明する基準を満たす場合は、コピーアクティビティを使用して、Azure Databricks Delta テーブルからシンクに直接コピーできます。設定が確認され、次の条件が満たされない場合は、コピーアクティビティの実行が失敗します。

シンクのリンクされたサービスは、Azure Blob Storage または Azure Data Lake Storage Gen2 です。アカウントの資格情報は Azure Databricks クラスター構成で事前に構成されている必要があります。詳細については、「前提条件」を参照してください。
シンクデータ形式は、次のように構成された Parquet、区切りテキスト、または Avro であり、ファイルではなくフォルダーを指しています。
- Parquet 形式の場合は、圧縮コーデックが none、snappy、または gzip です。
- 区切りテキスト形式の場合:
  - rowDelimiter は任意の 1 文字です。
  - compression には none、bzip2、gzip を指定できます。
  - encodingName UTF-7 はサポートされていません。
- Avro 形式の場合は、圧縮コーデックが none、deflate、または snappy です。
コピーアクティビティソースでは、additionalColumns が指定されていません。
区切りテキストにデータをコピーする場合、コピーアクティビティシンクでは、fileExtension は ".csv" である必要があります。
コピーアクティビティのマッピングで、型変換が有効になっていません。

例:

"activities":[
    {
        "name": "CopyFromDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Delta lake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureDatabricksDeltaLakeSource",
                "sqlReaderQuery": "SELECT * FROM events TIMESTAMP AS OF timestamp_expression"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Delta Lake からステージングコピーする

シンクのデータストアまたは形式が、前のセクションで説明した直接コピーの基準に適合しない場合は、中間の Azure Storage インスタンスを使用して組み込みのステージングコピーを有効にします。ステージングコピー機能はスループットも優れています。サービスが、Azure Databricks Delta Lake のデータをステージングストレージにエクスポートしてから、データをシンクにコピーし、最後にステージングストレージの一時データをクリーンアップします。ステージングを使用したデータのコピーの詳細は、「ステージングコピー」を参照してください。

この機能を使うには、中間ステージングとしてストレージアカウントを参照する Azure Blob Storage のリンクされたサービスまたは Azure Data Lake Storage Gen2 のリンクされたサービスを作成します。次に、コピーアクティビティに enableStaging プロパティと stagingSettings プロパティを指定します。

注意

ステージングストレージアカウントの資格情報は Azure Databricks クラスター構成で事前に構成されている必要があります。詳細については、「前提条件」を参照してください。

例:

"activities":[
    {
        "name": "CopyFromDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Delta lake input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureDatabricksDeltaLakeSource",
                "sqlReaderQuery": "SELECT * FROM events TIMESTAMP AS OF timestamp_expression"
            },
            "sink": {
                "type": "<sink type>"
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingStorage",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

シンクとしての Delta Lake

Azure Databricks Delta Lake にデータをコピーするために、コピーアクティビティの sink セクションでは次のプロパティがサポートされています。

プロパティ	内容	必須
type	コピーアクティビティのシンクの type プロパティ。AzureDatabricksDeltaLakeSink に設定します。	はい
preCopyScript	コピーアクティビティの毎回の実行で、データを Databricks Delta テーブルに書き込む前に実行する SQL クエリを指定します。例: `VACUUM eventsTable DRY RUN` このプロパティを使用して、事前に読み込まれたデータをクリーンアップしたり、TRUNCATE TABLE ステートメントまたは VACUUM ステートメントを追加したりできます。	いいえ
importSettings	デルタテーブルにデータを書き込むために使用される詳細設定。	いいえ
`importSettings` の下:
type	インポートコマンドの種類。AzureDatabricksDeltaLakeImportCommand に設定します。	Yes
dateFormat	日付形式の文字列を日付型に書式設定します。カスタム日付形式は datetime パターンの形式に従います。指定しない場合は、既定値の `yyyy-MM-dd` が使用されます。	いいえ
timestampFormat	タイムスタンプ形式の文字列をタイムスタンプ型に書式設定します。カスタム日付形式は datetime パターンの形式に従います。指定しない場合は、既定値の `yyyy-MM-dd'T'HH:mm:ss[.SSS][XXX]` が使用されます。	いいえ

Delta Lake に直接コピーする

ソースのデータストアと形式がこのセクションで説明する基準を満たす場合は、コピーアクティビティを使用して、ソースから Azure Databricks Delta Lake に直接コピーできます。設定が確認され、次の条件が満たされない場合は、コピーアクティビティの実行が失敗します。

ソースのリンクされたサービスは、Azure Blob Storage または Azure Data Lake Storage Gen2 です。アカウントの資格情報は Azure Databricks クラスター構成で事前に構成されている必要があります。詳細については、「前提条件」を参照してください。
ソースデータ形式は、次のように構成された Parquet、区切りテキスト、または Avro であり、ファイルではなくフォルダーを指しています。
- Parquet 形式の場合は、圧縮コーデックが none、snappy、または gzip です。
- 区切りテキスト形式の場合:
  - rowDelimiter は既定値または任意の 1 文字です。
  - compression には none、bzip2、gzip を指定できます。
  - encodingName UTF-7 はサポートされていません。
- Avro 形式の場合は、圧縮コーデックが none、deflate、または snappy です。
コピーアクティビティソース内:
- wildcardFileName にワイルドカード * のみが含まれ、? は含まれず、wildcardFolderName は指定されていません。
- prefix、modifiedDateTimeStart、modifiedDateTimeEnd、および enablePartitionDiscovery が指定されていない。
- additionalColumns が指定されていません。
コピーアクティビティのマッピングで、型変換が有効になっていません。

例:

"activities":[
    {
        "name": "CopyToDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Delta lake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureDatabricksDeltaLakeSink",
                "sqlReadrQuery": "VACUUM eventsTable DRY RUN"
            }
        }
    }
]

Delta Lake にステージングコピーする

ソースのデータストアまたは形式が、前のセクションで説明した直接コピーの基準に適合しない場合は、中間の Azure Storage インスタンスを使用して組み込みのステージングコピーを有効にします。ステージングコピー機能はスループットも優れています。サービスは、データ形式の要件を満たすようにデータをステージングストレージに自動的に変換し、そこから Delta Lake にデータを読み込みます。最後に、ストレージから一時データがクリーンアップされます。ステージングを使用したデータのコピーの詳細は、「ステージングコピー」を参照してください。

注意

例:

"activities":[
    {
        "name": "CopyToDeltaLake",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Delta lake output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureDatabricksDeltaLakeSink"
            },
            "enableStaging": true,
            "stagingSettings": {
                "linkedServiceName": {
                    "referenceName": "MyStagingBlob",
                    "type": "LinkedServiceReference"
                },
                "path": "mystagingpath"
            }
        }
    }
]

監視

他のコネクタと同様にコピーアクティビティのモニタリング体験が提供されます。さらに、Delta Lake との間のデータの読み込みが Azure Databricks クラスターで実行されているため、詳細なクラスターログを表示したり、パフォーマンスを監視したりすることができます。

Lookup アクティビティのプロパティ

プロパティの詳細については、ルックアップアクティビティに関する記事を参照してください。

ルックアップアクティビティでは、最大 1,000 行を返すことができます。結果セットにそれを超えるレコードが含まれている場合は、最初の 1,000 行が返されます。

コピーアクティビティによってソース、シンクとしてサポートされるデータストアの一覧については、サポートされるデータストアと形式の表を参照してください。

次の方法で共有

Azure Data Factory または Azure Synapse Analytics を使用して Azure Databricks Delta Lake との間でデータをコピーします

サポートされる機能

前提条件

クラスター構成の指定

はじめに

UI を使って Azure Databricks Delta Lake へのリンクサービスを作成する

コネクタの構成の詳細

リンクされたサービスのプロパティ

アクセストークン

システム割り当てマネージド ID 認証

ユーザー割り当てマネージド ID 認証

データセットのプロパティ

コピーアクティビティのプロパティ

ソースとしての Delta Lake

Delta Lake から直接コピーする

Delta Lake からステージングコピーする

シンクとしての Delta Lake

Delta Lake に直接コピーする

Delta Lake にステージングコピーする

監視

Lookup アクティビティのプロパティ

フィードバック

その他のリソース

次の方法で共有

Azure Data Factory または Azure Synapse Analytics を使用して Azure Databricks Delta Lake との間でデータをコピーします

サポートされる機能

前提条件

クラスター構成の指定

はじめに

UI を使って Azure Databricks Delta Lake へのリンク サービスを作成する

コネクタの構成の詳細

リンクされたサービスのプロパティ

アクセス トークン

システム割り当てマネージド ID 認証

ユーザー割り当てマネージド ID 認証

データセットのプロパティ

コピー アクティビティのプロパティ

ソースとしての Delta Lake

Delta Lake から直接コピーする

Delta Lake からステージング コピーする

シンクとしての Delta Lake

Delta Lake に直接コピーする

Delta Lake にステージング コピーする

監視

Lookup アクティビティのプロパティ

関連するコンテンツ

フィードバック

その他のリソース

UI を使って Azure Databricks Delta Lake へのリンクサービスを作成する

アクセストークン

コピーアクティビティのプロパティ

Delta Lake からステージングコピーする

Delta Lake にステージングコピーする