メインコンテンツまでスキップ
に最終更新

Microsoft Dynamics 365 コネクタ リファレンス

このリファレンスでは、Lakeflow Connect における Microsoft Dynamics 365 コネクターの認証、カーソルとスキーマの動作、サポートされている Dataverse データ型、スキーマ進化、およびパイプライン パラメーターについて説明します。

認証の問題​

Dynamics 365 コネクタは、Microsoft Entra ID (旧 Azure Active Directory) OAuth 認証を使用します。詳細については、 「コネクタはどのようにして D365 データにアクセスしますか?」を参照してください。

必須の認証フィールド​

D365 のUnity Catalog接続を作成するときは、次の点を指定します。

パラメーター

説明

例

tenant_id

Microsoft Entra ID テナント ID (ディレクトリ ID)

12345678-1234-1234-1234-123456789abc

client_id

Entra ID アプリのアプリケーション（クライアント）ID

87654321-4321-4321-4321-cba987654321

client_secret

Entra ID アプリ用に作成されたクライアント シークレット値

abc123~xyz789...

azure_storage_account_name

ADLS Gen2 ストレージ アカウントの名前

d365storage

azure_container_name

Synapse Linkがデータをエクスポートするコンテナ

d365-export

oauth_scope

Azure Storage アクセスの OAuth スコープ

https://storage.azure.com/.default

パラメーター

説明

例

tenant_id

Microsoft Entra ID テナント ID (ディレクトリ ID)

12345678-1234-1234-1234-123456789abc

client_id

Entra ID アプリのアプリケーション（クライアント）ID

87654321-4321-4321-4321-cba987654321

client_secret

Entra ID アプリ用に作成されたクライアント シークレット値

abc123~xyz789...

azure_storage_account_name

ADLS Gen2 ストレージ アカウントの名前

d365storage

azure_container_name

Synapse Linkがデータをエクスポートするコンテナ

d365-export

oauth_scope

Azure Storage アクセスの OAuth スコープ

https://storage.azure.com/.default

カーソルフィールド​

Dynamics 365 コネクタは、Azure Synapse Link の変更ログのversionnumberフィールドを増分取り込みのカーソルとして使用します。

カーソルの動作​

  • ソース : Synapse Link は変更をエクスポートするときにversionnumber値を自動的に生成します。
  • 形式 : 変更シーケンスを表す整数タイムスタンプ。
  • スコープ : テーブルごとのカーソル。各テーブルは独自のカーソル位置を維持します。
  • ストレージ : カーソルはパイプライン メタデータに保存され、ターゲット Delta テーブルには表示されません。

カーソルの要件​

増分取り込みを機能させるには:

  • Synapse Link はversionnumberフィールドを使用して変更ログをエクスポートする必要があります。
  • versionnumber すべての変更ログ ファイルに存在する必要があります。
  • Changelog フォルダーは、Synapse Link のタイムスタンプベースの命名規則に従う必要があります。

versionnumberがない場合、増分取り込みは失敗し、完全更新を実行する必要があります。

ソースエクスポート形式​

Azure Synapse Link は、Dataverse データを CSV または Parquet 形式で ADLS Gen2 にエクスポートできます。Dynamics 365 コネクタは両方をサポートしており、Synapse Link が書き込んだ形式を自動的に検出するため、パイプライン定義でエクスポート形式を指定する必要はありません。形式は、Synapse Link のセットアップによって決定されます。Parquet としてエクスポートするには Azure Synapse Analytics ワークスペースを接続する必要があります (Microsoft Dynamics 365 取り込み用の Parquet データソースを構成するを参照)。一方、CSV エクスポートでは標準のセットアップを使用します (Microsoft Dynamics 365 取り込み用のデータソースを構成するを参照)。

  • CSV ：Synapse Linkは、Azure Synapse Analyticsワークスペースを使用せずに、CSVファイルをADLS Gen2に直接書き込みます。
  • Parquet : Parquet の取り込みはベータ版です。Synapse Link は、各テーブルを <profileRoot>/deltalake/<tableName>/ 配下に Parquet 形式の Delta テーブルとして書き込みます。このパスには Azure Synapse Analytics ワークスペースと Apache Spark プールが必要です。Databricks では、大規模または高ボリュームのインスタンスに対してこのパスを推奨しています。

スキーマの検出​

Dynamics 365 コネクタは、Dataverse メタデータからテーブル スキーマを自動的に検出します。

発見プロセス​

パイプラインを作成するときは、次の点に注意してください。

  1. コネクタは、ADLS Gen2 から Synapse Link メタデータ ファイルを読み取ります。
  2. コネクタはメタデータ JSON ファイルからテーブル スキーマを抽出します。
  3. 列名、データ型、および NULL 値可能性はメタデータから推測されます。
  4. 検出されたスキーマを使用してターゲット テーブルが作成されます。

CSV取り込みのスキーマ進化​

備考

プレビュー

この機能は プライベート プレビュー段階です。試用については、Databricksの担当者にお問い合わせください。

スキーマ進化は、Azure Synapse Linkからの列の追加、削除、名前の変更など、ソースDataverseスキーマの変更に合わせて、ターゲットDeltaテーブルを自動的に同期し続けます。

前提条件​

  • スキーマ進化は、プライベートプレビュー期間中はオプトイン方式です。パイプラインでこれを有効にするには、Databricksの担当者にお問い合わせください。
  • 接続の Service Principal は、ADLS Gen2 ストレージ アカウントに対して Storage Blob Data Contributor ロールを持っている必要があります。スキーマ進化はスキーマチェックポイントをストレージに書き戻すため、読み取り権限のみを付与するロールでは機能しません。読み取り専用接続は、書き込みアクセスの付与またはスキーマ進化の無効化を求めるエラーで失敗します。接続のセットアップについては、「Dynamics 365 接続の作成」を参照してください。

スキーマの変更が適用される仕組み​

パイプラインの更新は、その期間中、単一の固定スキーマで読み取りを行うため、コネクタは変更されたスキーマを反映させるために新しい更新を起動します。ソースにおける各スキーマの変更は、次のように適用されます。

  1. この更新では、スキーマが変更される時点までのエクスポートされたデータを取り込み（変更点は含みません）、commitします。
  2. 更新はキャンセルされ、コネクタは新しいスキーマを記録します。
  3. 新しい更新はスキーマが変更された時点から開始され、そこから新しいスキーマで継続されます。

エクスポート履歴内のスキーマ変更ごとにこのサイクルが繰り返されるため、複数のスキーマ変更にまたがるバックログを処理するには、複数回の更新が必要となります。コネクターが自動的に更新を再開するため、操作は不要です。

再起動は、データが重複したり消失したりしないように設計されています。各更新は変更前に読み取ったデータのみをcommitし、次の更新は前回の停止地点から再開されます。

注記

これらの再起動は予想されるため、ソーススキーマの変更後にキャンセルされた更新は失敗とは言えません。パイプラインはスキーマ変更イベントを報告し、影響を受けたテーブルと新しいカラムの名前を報告します。

サポートされている変更とサポートされていない変更​

スキーマ進化は、以下のソーススキーマの変更を処理します。

変更

サポートされています

列の追加

サポートされています

列の削除

サポートされています

列の名前変更

サポートされています

テーブルの追加

サポートされています

テーブルの削除

サポートされています

列を削除し、後で同じ名前の列を再追加すること

サポートされていない

テーブルを削除し、後で同じ名前のテーブルを再追加する場合

サポートされていない

変更

サポートされています

列の追加

サポートされています

列の削除

サポートされています

列の名前変更

サポートされています

テーブルの追加

サポートされています

テーブルの削除

サポートされています

列を削除し、後で同じ名前の列を再追加すること

サポートされていない

テーブルを削除し、後で同じ名前のテーブルを再追加する場合

サポートされていない

サポートされているデータバースデータ型​

Dynamics 365 コネクタは、Dataverse データ型を Delta Lake データ型にマッピングします。

データ型のマッピング​

データバース型

Delta Lake型

注

String （1行）

STRING

メタデータとして保存される最大長

String （複数行）

STRING

Integer （整数）

INTEGER

BigInt

LONG

Decimal

DECIMAL

精度とスケールを維持

Double （浮動小数点）

DOUBLE

Money

DECIMAL(19,4)

小数点以下4桁の10進数として保存されます

Boolean （はい／いいえ）

BOOLEAN

DateTime

TIMESTAMP

タイムゾーン情報は保存されます

Date

DATE

Time

TIMESTAMP

SparkにはネイティブのTime型がないため、コネクタはTimeを TIMESTAMP

Uniqueidentifier (GUID)

STRING

文字列表現として保存

Lookup

STRING

文字列として保存された外部キー GUID

Picklist （オプションセット）

INTEGER

ラベルではなく整数値

Multi-select Picklist

STRING

カンマ区切りの整数値

Image

STRING

バイナリデータではなくURLまたはメタデータ

File

STRING

メタデータのみ、ファイルの内容は含まれません

データバース型

Delta Lake型

注

String （1行）

STRING

メタデータとして保存される最大長

String （複数行）

STRING

Integer （整数）

INTEGER

BigInt

LONG

Decimal

DECIMAL

精度とスケールを維持

Double （浮動小数点）

DOUBLE

Money

DECIMAL(19,4)

小数点以下4桁の10進数として保存されます

Boolean （はい／いいえ）

BOOLEAN

DateTime

TIMESTAMP

タイムゾーン情報は保存されます

Date

DATE

Time

TIMESTAMP

SparkにはネイティブのTime型がないため、コネクタはTimeを TIMESTAMP

Uniqueidentifier (GUID)

STRING

文字列表現として保存

Lookup

STRING

文字列として保存された外部キー GUID

Picklist （オプションセット）

INTEGER

ラベルではなく整数値

Multi-select Picklist

STRING

カンマ区切りの整数値

Image

STRING

バイナリデータではなくURLまたはメタデータ

File

STRING

メタデータのみ、ファイルの内容は含まれません

複雑なデータ型​

一部の Dataverse タイプでは特別な処理が必要です。

データバース型

取り込み形式：

対処方法

Picklist （オプションセット）

整数コード

OptionSetMetadata テーブルと結合するか、参照マッピングテーブルを維持してラベルを取得します

Lookup

GUID 文字列

参照テーブルと結合して関連データを取得します

Multi-select Picklist

カンマ区切りの整数文字列（例： "1,3,5"

文字列を解析して個々の値を抽出します

データバース型

取り込み形式：

対処方法

Picklist （オプションセット）

整数コード

OptionSetMetadata テーブルと結合するか、参照マッピングテーブルを維持してラベルを取得します

Lookup

GUID 文字列

参照テーブルと結合して関連データを取得します

Multi-select Picklist

カンマ区切りの整数文字列（例： "1,3,5"

文字列を解析して個々の値を抽出します

次の例では、複数選択オプションセットを解析し、まずカンマ区切りの値を配列に分割してから、それらを個別の行に展開（explode）します：

SQL
-- Split comma-separated values into array
SELECT
  accountid,
  accountname,
  SPLIT(industrycodes, ',') AS industry_array
FROM main.d365_data.account;

-- Explode into separate rows
SELECT
  accountid,
  accountname,
  CAST(code AS INT) AS industry_code
FROM main.d365_data.account
LATERAL VIEW EXPLODE(SPLIT(industrycodes, ',')) AS code;

APIバージョンの互換性​

Dynamics 365 コネクタは次のものと互換性があります。

  • Dataverse API : バージョン 9.2 以降
  • Azure Synapse Link for Dataverse : バージョン 1.0 以降
  • Azure Storage REST API : バージョン 2021-08-06 以降
  • Microsoft Entra ID : OAuth 2.0 クライアント認証情報フロー
注記

古いバージョンの API も動作する可能性がありますが、公式にはサポートされていません。最高の互換性を得るために、D365 および Azure サービスを最新の状態に保ってください。

増分摂取行動​

コネクタが検出された各変更をどのように適用するかは、パイプラインに対して選択した SCD タイプによって異なり、削除が Databricks に反映されるかどうかは Synapse Link の構成によって異なります。

変更検出​

このコネクタは、Synapse Link の変更ログからすべての変更を検出します。変更ログにレコードが存在する場合は挿入、変更された versionnumber は更新、削除マーカーは削除として扱います。削除マーカーは、Synapse Link が削除をエクスポートするように構成されている場合にのみ表示されます。

SCDタイプ1の行動​

SCD タイプ 1パイプラインの場合、レコードは履歴を保持せずにその場で更新されます。更新は主キーに基づいて既存の行を上書きし、削除は行を削除します (削除追跡が有効になっている場合)。

ターゲットテーブルをクエリーすると、レコードごとに1行が返され、最新の状態のみが反映されます：

SQL
SELECT * FROM main.d365_data.account ORDER BY accountid;

-- Result: Latest state only
-- accountid | accountname | modifiedon
-- 123       | Acme Corp   | 2025-12-03 10:00:00
-- 456       | TechCo      | 2025-12-03 09:30:00

SCDタイプ2の行動​

SCD タイプ 2パイプラインの場合、すべての変更は新しい行バージョンとして保持されます。コネクタは、バージョン履歴を追跡するために__START_AT 、 __END_AT 、および__CURRENT列を追加します。

ターゲットテーブルのクエリーを実行すると、各レコードのすべてのバージョンが返されます。__START_AT と __END_AT は各バージョンが有効であった期間を制限し、アクティブなバージョンには true に設定された NULL __END_AT および __CURRENT があります：

SQL
SELECT * FROM main.d365_data.account ORDER BY accountid, __START_AT;

-- Result: All historical versions
-- accountid | accountname | __START_AT          | __END_AT            | __CURRENT
-- 123       | Acme Inc    | 2025-11-01 08:00:00 | 2025-12-03 10:00:00 | false
-- 123       | Acme Corp   | 2025-12-03 10:00:00 | NULL                | true
-- 456       | TechCo      | 2025-12-01 14:00:00 | NULL                | true
注記

Synapse Link が Parquet としてエクスポートする場合、古い Delta チェックポイントは定期的に新しいものに圧縮されます。これにより、一部の履歴レコードバージョンが処理に使用できなくなる可能性があります。SCD タイプ 2 パイプラインの場合、履歴が不完全になる可能性がありますが、データ損失は発生しません。パイプラインは常に各レコードの最新のスナップショットを正しく反映しており、一部の中間バージョンのみが欠落する可能性があります。履歴が不完全になる可能性を減らすため、Databricks では 24 時間に 1 回よりも頻繁にパイプラインを実行することを推奨しています。これにより、圧縮によって履歴の変更が削除される可能性のある期間が短縮されます。

削除処理​

削除の処理は Synapse Link の構成によって異なります。

  • ハード削除 : Synapse Link が削除レコードをエクスポートする場合、コネクタは削除されたレコードを削除するか (SCD タイプ 1)、削除されたレコードをマークします (SCD タイプ 2)。
  • 削除の追跡なし : Synapse Link が削除をエクスポートしない場合、削除されたレコードは完全更新を実行するまでターゲット テーブルに残ります。

正確な削除追跡が必要な場合は、Synapse Link 構成で削除がエクスポートされることを確認してください。

パイプラインある​

D365 インジェスト パイプラインを作成するときは、次の点を指定します。

必須​

パイプラインを実行するには、これらのパラメーターを指定する必要があります。

パラメーター

Type

説明

例

channel

String

でなければなりません PREVIEW

"PREVIEW"

connection_name

String

Unity Catalog接続の名前

"d365_connection"

source_schema

String

Synapse Link の論理スキーマ名。通常は objects

"objects"

source_table

String

D365 テーブルの論理名（table オブジェクトごとに 1 つのエントリ）

"account"

destination_catalog

String

ターゲットUnity Catalogカタログ

"main"

destination_schema

String

ターゲットUnity Catalogスキーマ

"d365_data"

scd_type

String

SCD_TYPE_1 または SCD_TYPE_2

"SCD_TYPE_2"

パラメーター

Type

説明

例

channel

String

でなければなりません PREVIEW

"PREVIEW"

connection_name

String

Unity Catalog接続の名前

"d365_connection"

source_schema

String

Synapse Link の論理スキーマ名。通常は objects

"objects"

source_table

String

D365 テーブルの論理名（table オブジェクトごとに 1 つのエントリ）

"account"

destination_catalog

String

ターゲットUnity Catalogカタログ

"main"

destination_schema

String

ターゲットUnity Catalogスキーマ

"d365_data"

scd_type

String

SCD_TYPE_1 または SCD_TYPE_2

"SCD_TYPE_2"

オプション​

これらのパラメーターは、パイプラインの作成時に任意で設定できます。

パラメーター

Type

説明

例

table_configuration

オブジェクト

列の選択などのテーブルごとの設定

列の選択を参照

パラメーター

Type

説明

例

table_configuration

オブジェクト

列の選択などのテーブルごとの設定

列の選択を参照

パイプライン構成の例​

これは、Python SDK を使用した完全なパイプライン構成の例です。

Python
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import IngestionPipelineDefinition

w = WorkspaceClient()

pipeline = w.pipelines.create(
    name="d365_comprehensive_ingestion",
    ingestion_definition=IngestionPipelineDefinition(
        channel="PREVIEW",
        connection_name="d365_connection",
        source_schema="objects",
        source_table="account",
        destination_catalog="main",
        destination_schema="d365_sales",
        scd_type="SCD_TYPE_2",
        table_configuration={
            &quot;account&quot;: {
                &quot;columns&quot;: [
                    &quot;accountid&quot;,
                    &quot;accountnumber&quot;,
                    &quot;name&quot;,
                    &quot;emailaddress1&quot;,
                    &quot;telephone1&quot;
                ]
            }
        }
    )
)

テーブルの論理名の検索​

source_table問題のテーブル論理名を特定するには:

  1. Power Apps メーカーポータル : テーブル に移動し、 論理名 列を表示します。
  2. Dataverse API : https://yourorg.api.crm.dynamics.com/api/data/v9.2/EntityDefinitionsを使用してメタデータをクエリします。
  3. ADLS Gen2 ストレージ : Synapse Link コンテナー内のフォルダーを一覧表示します (フォルダー名は論理名と一致します)。
ヒント

パイプライン構成では小文字の論理名を使用します (たとえば、 "account"ではなく"Account" )。コネクタでは大文字と小文字が区別されます。

パフォーマンスチューニング​

Synapse Link自体がエクスポートを制御するため、コネクタで可能なチューニングは制限されています。制御できるのは、各パイプラインが移動するデータ量と、パイプライン間でテーブルをどのように分散させるかです。

列の選択​

必要な列のみを選択することで、ADLS Gen2からのデータ転送、Delta Lakeのストレージコスト、およびクエリーの処理時間を削減できます。構成の詳細については、 列の選択 を参照してください。

テーブルのグループ化​

動作が類似しているテーブルをグループ化し、パイプラインのスケジュールがその中のすべてに適するようにします。管理を容易にするために関連テーブルをまとめ、更新パターンが類似しているテーブルをまとめることで、各パイプラインのスケジュールを変更量に合わせて調整できます。高負荷のテーブルには独自のパイプラインを割り当て、単一の大きなテーブルが他の処理を遅延させないようにします。

各パイプラインは 250 テーブルに制限されます。より大規模な環境の場合は、複数のパイプラインを作成します。

トラブルシューティング​

Dynamics 365 コネクタを使用する場合の一般的な問題と解決策については、 Microsoft Dynamics 365 取り込みのトラブルシューティング」を参照してください。

このページの見出し