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.

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 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 doivent être correctement configurées et synchronisées avant d'apparaître dans Azure Synapse Link. Laissez-leur le temps de devenir disponibles.

Configurer Azure Synapse Link

Dans cette étape, vous utiliserez le Synapse Link for Dataverse vers Azure Data Lake pour choisir les tables que vous souhaitez ingérer.

remarque

Synapse Link for Dataverse to Azure Data Lake remplace le service anciennement connu sous le nom de Export data to Azure Data Lake Storage Gen2. Malgré la dénomination, cette fonctionnalité n'utilise pas ni ne dépend d'Azure Synapse Analytics ; il s'agit d'un service d'exportation continue de Dataverse vers ADLS Gen 2.

remarque

Par default, Microsoft prend actuellement en charge jusqu'à 1 000 tables par profil Synapse Link. Si votre application a plus de tables sélectionnées, vous devrez créer plusieurs profils.

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

2. Cliquez sur Nouveau Link . Dataverse remplira automatiquement vos abonnements actifs du même tenant. Sélectionnez l'abonnement approprié dans la liste déroulante.

3. Ne sélectionnez pas la case à cocher **Connecter à votre Azure Synapse Analytics Workspace**. (Les données doivent atterrir en CSV car le connecteur ne prend pas actuellement en charge Parquet.)

remarque

Pour utiliser ce connecteur, vous ne pouvez pas ajouter de tables Dataverse à un compte de stockage existant déjà lié à un profil Synapse Link différent. Vous devez avoir accès à un abonnement Azure non lié afin de pouvoir créer un nouveau profil Synapse Link.

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 Ajouter uniquement et Partition 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 .

8. Après avoir enregistré vos modifications, la synchronisation initiale de Synapse Link devrait commencer.

remarque

Pour les utilisateurs F&O, cette synchronisation initiale peut prendre des heures pour les tables volumineuses de centaines de gigaoctets. Toutefois, si la synchronisation initiale d'une entité prend trop de temps, vous pouvez l'accélérer en créant un index sur la table via l'application F&O.

• Accédez à la table que vous souhaitez indexer dans l'environnement F&O.
• Créez une extension pour la table.
• Dans l'extension de table, définissez un nouvel index.
• Ajoutez les champs que vous souhaitez inclure dans l'index. (Ceci aide à accélérer la recherche dans la base de données basée sur ces champs.)
• 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

Utilisez l’interface utilisateur

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é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.

Créer un pipeline Dynamics 365

À cette étape, vous configurerez le pipeline d'ingestion. Chaque table ingérée reçoit une table de streaming correspondante avec le même nom dans la destination.

Il existe deux options pour la création du pipeline d'ingestion :

• Utilisez un Notebook
• Utilisez la CLI Databricks

Les deux approches effectuent des appels d'API à un service Databricks qui crée le pipeline.

Notebook:

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.

CLI:

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

Cellule 1

Ne modifiez pas cette cellule.

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

Ne modifiez pas le canal de distribution PREVIEW dans le code ci-dessous.

Si vous souhaitez ingérer toutes les tables synchronisées par votre Azure Synapse Link, utilisez la spécification au niveau du schéma. (Notez, cependant, que Databricks ne recommande pas d'ajouter plus de 250 tables par pipeline.) Si vous ne souhaitez ingérer que des tables spécifiques, utilisez la spécification au niveau de la table.

Assurez-vous que le nom de la table source correspond au nom de la table qui apparaît dans la colonne Nom de la page de gestion Synapse Link.

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 : 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 : sélection et désélection au niveau de la colonne

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)