Configurer la source de données pour l’ingestion Microsoft Dynamics 365

Découvrez comment configurer Microsoft Dynamics 365 comme source de données pour l'ingestion dans Databricks à l'aide de Lakeflow Connect.

remarque

Cette page couvre le workflow d’exportation CSV, qui n’utilise pas de workspace Azure Synapse Analytics. Pour exporter en tant que tables Delta au format Parquet en utilisant plutôt un workspace Azure Synapse Analytics, consultez Configurer une source de données Parquet pour l’ingestion Microsoft Dynamics 365. Databricks recommande le workflow Parquet pour les instances volumineuses ou à haut débit, car il offre de meilleures performances et une meilleure stabilité lors de la montée en charge.

Pour des informations sur la manière dont le connecteur accède à vos données source, consultez Comment le connecteur accède-t-il aux données D365 ? Pour obtenir la liste des applications Dataverse prises en charge, consultez Quelles applications Dynamics 365 sont prises en charge ?

Prérequis

Avant de configurer la source de données Dynamics 365, vous devez disposer de :

• Un abonnement Azure actif avec les autorisations nécessaires pour créer des Ressources.
• Un environnement Microsoft Dynamics 365 avec accès administrateur.
• Un environnement Dataverse associé à votre instance Dynamics 365.
• Autorisations d'administrateur de Workspace ou d'administrateur de métastore dans Databricks.
• Permissions pour créer et configurer Azure Synapse Link dans votre environnement Dataverse.
• Un abonnement Azure avec un compte de stockage qui n’est pas déjà associé à un autre profil Synapse Link. Vous ne pouvez pas ajouter de tables Dataverse à un compte de stockage associé à un profil différent ; vous devez créer un nouveau profil Synapse Link.
• Un compte de stockage ADLS Gen2 (ou les autorisations pour en créer un).
• Autorisations pour créer et configurer des applications Microsoft Entra ID.
• Dataverse API v9,2 ou ultérieure.
• Version 2021-08-06 de l’API REST de Stockage Azure.
• Azure Synapse Link pour Dataverse version 1.0 ou ultérieure.

Configurez les entités virtuelles ou les tables directes (facultatif)

Les entités virtuelles et les tables directes rendent les données provenant de sources non Dataverse (telles que Dynamics 365 Finance & Opérations) disponibles dans Dataverse sans copier les données. Pour les sources non Dataverse, vous devez configurer des entités virtuelles ou des tables directes avant de configurer Azure Synapse Link.

Pour configurer les entités virtuelles :

1. Dans **Power Apps**, accédez à la page **Environnements**, puis cliquez sur **applications Dynamics 365**.

2. Pour lier les entités F&O en tant qu'entités virtuelles dans Dataverse, installez la solution Finance and Operations Virtual Entity .

3. Configurez l'autorisation Service To Service (S2S) entre Dataverse et votre application F&O. Cela permet à Dataverse de communiquer avec votre application. Pour plus de détails, consultez la documentation Microsoft Configurer les entités virtuelles Dataverse.

4. Pour chaque entité virtuelle que vous souhaitez ingérer, activez **Suivi des modifications** sous **Propriétés avancées**.

5. Par default, la solution d'entité virtuelle F&O expose certaines entités virtuelles par default dans la liste des tables Dataverse. Cependant, vous pouvez exposer des entités supplémentaires manuellement :

   1. Accédez à la page Paramètres avancés de votre environnement Dataverse.
   2. Cliquez sur l'icône de filtre en haut à droite pour accéder à la recherche avancée.
   3. Sélectionnez Entités financières et d'opérations disponibles dans le menu déroulant, puis cliquez sur Résultats .
   4. Sélectionner l'entité virtuelle que vous souhaitez exposer.
   5. Sur la page Entity Admin , passez Visible à True , puis cliquez sur Save and Close .

Vous pouvez maintenant voir l’entité dans la liste des tables Dataverse avec un nom qui start avec mserp_.

important

Les entités virtuelles et les tables directes n’apparaissent dans Azure Synapse Link qu’une fois que Dataverse a terminé leur synchronisation. Cette opération prend généralement jusqu’à 15 minutes, mais peut prendre jusqu’à 30 minutes. Si des tables sont manquantes après 30 minutes, consultez Les entités virtuelles n’apparaissent pas dans la découverte de schéma.

Configurer Azure Synapse Link

Dans cette étape, vous utiliserez Synapse Link for Dataverse to Azure Data Lake pour choisir les tables que vous souhaitez ingérer. Ce service remplace celui anciennement appelé Export data to Azure Data Lake Storage Gen2. Malgré son nom, il n'utilise pas et ne dépend pas d'Azure Synapse Analytics. Il s'agit d'un service d'exportation continue de Dataverse vers ADLS Gen2.

1. Dans le **Portail Power Apps**, cliquez sur **Analyser**, puis **Link à Azure Synapse**.

2. Cliquez sur New Link . Dataverse renseigne automatiquement vos abonnements actifs à partir du même tenant. Sélectionnez l’abonnement approprié dans la liste déroulante.

3. Ne cochez pas la case Connect to your Azure Synapse Analytics Workspace . Les données arrivent au format CSV directement dans votre compte de stockage ADLS Gen2, et ce workflow ne nécessite pas de workspace Azure Synapse Analytics.

4. Sur la page de création Synapse Link , cliquez sur Avancé . Puis, basculez Afficher les paramètres de configuration avancés .

5. Activez ou désactivez **Activer la structure de dossiers de mise à jour incrémentielle** et définissez l'intervalle de mise à jour souhaité du Synapse Link. Le minimum est de 5 minutes. Cet intervalle s'applique à toutes les tables incluses dans ce Synapse Link. (Vous définirez un planning pour votre pipeline Databricks dans une étape distincte.)

6. Sélectionnez les tables que vous souhaitez synchroniser, en laissant les paramètres Append only et Partition sur « default ».

   • Si vous ingérez à partir d'une application native Dataverse, sélectionnez les tables Dataverse pertinentes directement depuis la section **Dataverse**.
   • Si vous ingérez depuis F&O, vous pouvez sélectionner des tables directes dans la section D365 Finance & Opérations ou des entités virtuelles dans la section Dataverse (préfixe mserp_). Pour plus d'informations sur les entités virtuelles, consultez l'étape 1.

7. Cliquez sur Enregistrer . La synchronisation initiale de Synapse Link commence.

   Pour les utilisateurs de F&O, cette synchronisation initiale peut prendre des heures pour les grandes tables contenant des centaines de gigaoctets.

remarque

Si la synchronisation initiale d’une entité F&O prend trop de temps, vous pouvez l’accélérer en créant un index sur la table dans l’application F&O :

1. Accédez à la table que vous souhaitez indexer dans l'environnement F&O.
2. Créez une extension pour la table.
3. Dans l'extension de table, définissez un nouvel index.
4. Ajoutez les champs que vous souhaitez inclure dans l’index, ce qui accélère les recherches dans la base de données sur ces champs.
5. Enregistrez et déployez les modifications dans votre environnement F&O.

Créez une application Entra ID pour l'ingestion

Au cours de cette étape, vous collecterez les informations Entra ID nécessaires pour créer une connexion Unity Catalog qui prend en charge l'ingestion dans Databricks.

1. Collectez l'**ID de tenant** de votre tenant Entra ID (portal.azure.com >> **Microsoft Entra ID** >> **Overview tab** >> **ID de tenant**, listé dans le panneau de droite).

2. Lorsque vous créez un Azure Synapse Link, Azure Synapse crée un conteneur ADLS pour synchroniser les tables sélectionnées. Localisez le **nom du conteneur ADLS** en visitant la **page d'administration** de Synapse Link.

3. Collectez les identifiants d'accès pour le conteneur ADLS.

   1. Créer une application Microsoft Entra ID, si vous n'en avez pas déjà une.
   2. Collectez le secret client .
   3. Collectez l' ID d'application (portal.azure.com >> Microsoft Entra ID >> Gérer >> Enregistrements d'applications ).

4. Accordez à l'application Entra ID l'accès au conteneur ADLS, si ce n'est pas déjà fait.

remarque

Assurez-vous que votre application Entra ID a accès aux conteneurs ADLS associés à chaque profil Synapse Link. Si vous ingérez des données depuis plusieurs environnements ou applications, confirmez que l'application dispose d'attributions de rôles sur tous les conteneurs pertinents.

1. Accédez à Comptes de stockage Azure et sélectionnez votre conteneur ou compte de stockage. (Databricks recommande le niveau de conteneur pour maintenir les privilèges minimum.)
2. Cliquez sur **Contrôle d'accès (IAM)**, puis sur **Ajouter une attribution de rôle**.
3. Sélectionnez le rôle d’accès Contributeur aux données Blob du stockage → Lecture/Écriture/Suppression. Si votre organisation ne le permet pas, contactez votre équipe de compte Databricks.
4. Cliquez sur **Suivant**, puis sur **Sélectionner des membres**.
5. Choisissez Utilisateur, groupe ou Service Principal , puis Recherchez votre enregistrement d'application . (Si l'application n'est pas présente dans les résultats de recherche, vous pouvez explicitement entrer son ID d'objet dans la barre de recherche, puis appuyez sur Entrée ).
6. Cliquez sur Réviser + Attribuer .
7. Pour confirmer que les autorisations sont configurées correctement, vous pouvez vérifier le contrôle d'accès de votre conteneur.

Créer un pipeline Dynamics 365

Vous pouvez créer le pipeline dans l’interface utilisateur ou via l’API. L'assistant de l'interface utilisateur gère la connexion et le pipeline ensemble, tandis que le chemin via l'API les crée en deux étapes distinctes.

Utilisez l’interface utilisateur

L’assistant vous demande les identifiants d’application Entra ID et les détails de stockage que vous avez recueillis lors des étapes précédentes, puis crée la connexion et le pipeline ensemble.

1. Dans le menu de gauche, cliquez sur **Nouveau**, puis sur **Ajouter ou upload des données**.
2. Sur la page Ajouter des données , cliquez sur la vignette Dynamics 365 .
3. Suivez les instructions de l'assistant à partir de là.

Utilisez l'API

Créez d’abord la connexion, puis le pipeline qui l’utilise. Vous avez besoin du nom de connexion de la première étape pour définir le pipeline dans la seconde.

Étape 1 : créer une connexion Dynamics 365

Dans cette étape, vous allez créer une connexion Unity Catalog pour stocker en toute sécurité vos identifiants Dynamics 365 et commencer l'ingestion dans Databricks.

1. Dans votre Workspace, cliquez sur Catalogue .
2. Cliquez sur Connexion , puis cliquez sur Connexions .
3. Veuillez cliquer sur le bouton Créer une connexion .
4. Fournissez un nom de connexion unique, puis sélectionnez Dynamics 365 comme type de connexion .
5. Saisissez le secret client et l' identifiant client de l'application Entra ID créée à l'étape précédente. Ne modifiez pas la portée. Cliquez sur Suivant .
6. Saisissez le Nom du compte de stockage Azure , l' ID du tenant et le Nom du conteneur ADLS , puis cliquez sur Créer une connexion.
7. Notez le nom de la connexion.

Étape 2 : Créer le pipeline d’ingestion

Dans cette étape, vous allez configurer le pipeline d’ingestion. Chaque table ingérée obtient une table de streaming correspondante portant le même nom dans la destination. Vous pouvez utiliser un Notebook ou la CLI Databricks. Les deux approches effectuent des appels API vers un service Databricks qui crée le pipeline.

Utiliser un notebook

Le Template à la fin de cette page définit des fonctions d'assistance pour la création et la gestion du pipeline. La première cellule configure ces fonctions, et la seconde est celle où vous définissez votre propre pipeline.

1. Copiez le Template Notebook.
2. Exécutez la première cellule du Notebook sans la modifier.
3. Modifiez la deuxième cellule du Notebook avec les détails de votre pipeline (par exemple, la table à partir de laquelle vous souhaitez ingérer, l’endroit où vous souhaitez stocker les données, etc.).
4. Exécutez la deuxième cellule du Notebook template ; ceci exécute create_pipeline.
5. Vous pouvez exécuter list_pipeline pour afficher l'ID du pipeline et ses détails.
6. Vous pouvez exécuter edit_pipeline pour modifier la définition du pipeline.
7. Vous pouvez exécuter delete_pipeline pour supprimer le pipeline.

Utilisez la CLI Databricks

Pour créer le pipeline :

Bash

databricks pipelines create --json "<pipeline_definition OR json file path>"

Pour modifier le pipeline :

Bash

databricks pipelines update --json "<<pipeline_definition OR json file path>"

Pour obtenir la définition du pipeline :

Bash

databricks pipelines get "<your_pipeline_id>"

Pour supprimer le pipeline :

Bash

databricks pipelines delete "<your_pipeline_id>"

Pour plus d'informations, vous pouvez toujours exécuter :

Bash

databricks pipelines --help
databricks pipelines <create|update|get|delete|...> --help

Configurer des fonctionnalités supplémentaires (facultatif)

Le connecteur offre des fonctionnalités supplémentaires, telles que le SCD de type 2 pour le suivi de l'historique, la sélection et la désélection au niveau des colonnes. Voir Modèles courants pour les pipelines d'ingestion gérés.

Template de Notebook

Copiez les deux cellules dans un Notebook de votre workspace. La cellule 1 définit les fonctions d’assistance qui appellent l’API de pipeline, et la cellule 2 est l’endroit où vous définissez le pipeline que vous souhaitez créer.

Cellule 1 : configuration de l’API

Copiez cette cellule telle quelle et exécutez-la sans modification. Il définit create_pipeline, list_pipeline, edit_pipeline, delete_pipeline et les autres assistants appelés par la cellule 2.

Python

# DO NOT MODIFY

# This sets up the API utils for creating managed ingestion pipelines in Databricks.

import requests
import json

notebook_context = dbutils.notebook.entry_point.getDbutils().notebook().getContext()
api_token = notebook_context.apiToken().get()
workspace_url = notebook_context.apiUrl().get()
api_url = f"{workspace_url}/api/2.0/pipelines"

headers = {
    'Authorization': 'Bearer {}'.format(api_token),
    'Content-Type': 'application/json'
}

def check_response(response):
    if response.status_code == 200:
        print("Response from API:\n{}".format(json.dumps(response.json(), indent=2, sort_keys=False)))
    else:
        print(f"Failed to retrieve data: error_code={response.status_code}, error_message={response.json().get('message', response.text)}")

def create_pipeline(pipeline_definition: str):
  response = requests.post(url=api_url, headers=headers, data=pipeline_definition)
  check_response(response)

def edit_pipeline(id: str, pipeline_definition: str):
  response = requests.put(url=f"{api_url}/{id}", headers=headers, data=pipeline_definition)
  check_response(response)

def delete_pipeline(id: str):
  response = requests.delete(url=f"{api_url}/{id}", headers=headers)
  check_response(response)

def list_pipeline(filter: str):
  body = "" if len(filter) == 0 else f"""{{"filter": "{filter}"}}"""
  response = requests.get(url=api_url, headers=headers, data=body)
  check_response(response)

def get_pipeline(id: str):
  response = requests.get(url=f"{api_url}/{id}", headers=headers)
  check_response(response)

def start_pipeline(id: str, full_refresh: bool=False):
  body = f"""
  {{
    "full_refresh": {str(full_refresh).lower()},
    "validate_only": false,
    "cause": "API_CALL"
  }}
  """
  response = requests.post(url=f"{api_url}/{id}/updates", headers=headers, data=body)
  check_response(response)

def stop_pipeline(id: str):
  print("cannot stop pipeline")

Cellule 2 : définition du pipeline

Choisissez l'une des deux options ci-dessous, selon la quantité de données Synapse Link que vous souhaitez ingérer :

• Option A, spécification au niveau du schéma : ingère chaque table synchronisée par votre Azure Synapse Link. Databricks ne recommande pas plus de 250 tables par pipeline ; si votre Synapse Link en synchronise davantage, répartissez les tables sur plusieurs pipelines.
• Option B, spécification au niveau de la table : ingère uniquement les tables que vous nommez. Chaque valeur source_table doit correspondre au nom de la table dans la colonne Nom de la page Gérer de Synapse Link.

Remplacez les valeurs des espaces réservés par les vôtres, mais laissez "channel": "PREVIEW" tel quel.

Python

# Option A: schema-level spec
pipeline_spec = """
{
 "name": "<YOUR_PIPELINE_NAME>",
 "ingestion_definition": {
     "connection_name": "<YOUR_CONNECTION_NAME>",
     "objects": [
        {
          "schema": {
            "source_schema": "objects",
            "destination_catalog": "<YOUR_DATABRICKS_CATALOG>",
            "destination_schema": "<YOUR_DATABRICKS_SCHEMA>"
          }
        }
      ]
 },
 "channel": "PREVIEW"
}
"""

create_pipeline(pipeline_spec)

Python

# Option B: table-level spec
pipeline_spec = """
{
 "name": "<YOUR_PIPELINE_NAME>",
 "ingestion_definition": {
     "connection_name": "<YOUR_CONNECTION_NAME>",
     "objects": [
        {
          "table": {
            "source_schema": "objects",
            "source_table": "<YOUR_F_AND_O_TABLE_NAME>",
            "destination_catalog": "<YOUR_DATABRICKS_CATALOG>",
            "destination_schema": "<YOUR_DATABRICKS_SCHEMA>"
          }
        }
      ]
 },
 "channel": "PREVIEW"
}
"""

create_pipeline(pipeline_spec)

Exemple : suivre l’historique avec le SCD de type 2

Par default, l'API utilise le SCD de type 1. Cela signifie qu'il écrase les données dans la destination si elles sont modifiées dans la source. Si vous préférez conserver les données historiques et utiliser le SCD de type 2, spécifiez-le dans la configuration. Par exemple :

Python

# Schema-level spec with SCD type 2
pipeline_spec = """
{
 "name": "<YOUR_PIPELINE_NAME>",
 "ingestion_definition": {
     "connection_name": "<YOUR_CONNECTION_NAME>",
     "objects": [
        {
          "schema": {
            "source_schema": "objects",
            "destination_catalog": "<YOUR_DATABRICKS_CATALOG>",
            "destination_schema": "<YOUR_DATABRICKS_SCHEMA>",
        "table_configuration": {
              "scd_type": "SCD_TYPE_2"
            }
          }
        }
      ]
 },
 "channel": "PREVIEW"
}
"""

create_pipeline(pipeline_spec)

Python

# Table-level spec with SCD type 2
pipeline_spec = """
{
 "name": "<YOUR_PIPELINE_NAME>",
 "ingestion_definition": {
     "connection_name": "<YOUR_CONNECTION_NAME>",
     "objects": [
        {
          "table": {
            "source_schema": "objects",
            "source_table": "<YOUR_F_AND_O_TABLE_NAME>",
            "destination_catalog": "<YOUR_DATABRICKS_CATALOG>",
            "destination_schema": "<YOUR_DATABRICKS_SCHEMA>",
        "table_configuration": {
              "scd_type": "SCD_TYPE_2"
            }
          }
        }
      ]
 },
 "channel": "PREVIEW"
}
"""

create_pipeline(pipeline_spec)

Exemple : inclure ou exclure des colonnes spécifiques

Par default, l'API ingère toutes les colonnes de la table sélectionnée. Cependant, vous pouvez choisir d'inclure ou d'exclure des colonnes spécifiques. Par exemple :

Python

# Table spec with included and excluded columns.
pipeline_spec = """
{
 "name": "<YOUR_PIPELINE_NAME>",
 "ingestion_definition": {
     "connection_name": "<YOUR_CONNECTON_NAME>",
     "objects": [
        {
          "table": {
            "source_schema": "objects",
            "source_table": "<YOUR_F_AND_O_TABLE_NAME>",
            "destination_catalog": "<YOUR_DATABRICKS_CATALOG>",
            "destination_schema": "<YOUR_DATABRICKS_SCHEMA>",
        "table_configuration": {
          "include_columns": ["<COLUMN_A>", "<COLUMN_B>", "<COLUMN_C>"]
            }
          }
        }
      ]
 },
 "channel": "PREVIEW"
}
"""

create_pipeline(pipeline_spec)