Konfigurieren der Datenquelle für die Erfassung von Microsoft Dynamics 365

Erfahren Sie, wie Sie Microsoft Dynamics 365 als Datenquelle für die Aufnahme in Azure Databricks mithilfe von Lakeflow Connect einrichten.

Informationen dazu, wie der Connector auf Ihre Quelldaten zugreift, finden Sie unter Wie greift der Connector auf D365-Daten zu?. Eine Liste der unterstützten Dataverse-Anwendungen finden Sie unter Welche Dynamics 365-Anwendungen werden unterstützt?.

Voraussetzungen

Bevor Sie die Dynamics 365 Datenquelle konfigurieren, müssen Sie folgendes haben:

• Ein aktives Azure-Abonnement mit Berechtigungen zum Erstellen von Ressourcen.
• Eine Microsoft Dynamics 365-Umgebung mit Administratorzugriff.
• Eine Dataverse-Umgebung, die Ihrer Dynamics 365 Instanz zugeordnet ist.
• Arbeitsbereichsadministrator- oder Metastore-Administratorberechtigungen in Azure Databricks.
• Berechtigungen zum Erstellen und Konfigurieren des Azure Synapse-Links in Ihrer Dataverse-Umgebung.
• Ein Azure-Abonnement mit einem Speicherkonto, das nicht bereits mit einem anderen Synapse Link-Profil verknüpft ist. Man kann Dataverse-Tabellen nicht zu einem Speicherkonto hinzufügen, das mit einem anderen Profil verknüpft ist; Sie müssen ein neues Synapse Link-Profil erstellen.
• Ein ADLS Gen2-Speicherkonto (oder Berechtigungen zum Erstellen eines Speicherkontos).
• Berechtigungen zum Erstellen und Konfigurieren von Microsoft Entra-ID-Anwendungen.
• Dataverse-API v9.2 oder höher.
• Azure Storage REST API Version 2021-08-06.
• Azure Synapse Link für Dataverse, Version 1.0 oder höher.

Konfigurieren virtueller Entitäten oder direkter Tabellen (optional)

Virtuelle Entitäten und direkte Tabellen machen Daten aus Quellen außerhalb von Dataverse (wie z. B. Dynamics 365 Finance & Operations) in Dataverse verfügbar, ohne die Daten zu kopieren. Für Nicht-Dataverse-Quellen müssen Sie virtuelle Entitäten oder direkte Tabellen konfigurieren, bevor Sie Azure Synapse Link einrichten.

So konfigurieren Sie virtuelle Entitäten:

1. Wechseln Sie in Power Apps zur Seite "Umgebungen ", und klicken Sie dann auf Dynamics 365-Apps.

2. Um F&O-Entitäten als virtuelle Entitäten in Dataverse zu verknüpfen, installieren Sie die Lösung "Finance and Operations Virtual Entity" .

3. Richten Sie die Dienst-zu-Dienst-Autorisierung (S2S) zwischen Dataverse und Ihrer F&O-Anwendung ein. Auf diese Weise kann Dataverse mit Ihrer Anwendung kommunizieren. Ausführliche Informationen finden Sie in der Microsoft-Dokumentation Konfigurieren virtueller Entitäten von Dataverse.

4. Aktivieren Sie für jede virtuelle Entität, die Sie erfassen möchten, Änderungen nachverfolgen unter Erweiterte Eigenschaften.

5. Standardmäßig macht die F&O Virtual Entity-Lösung einige virtuelle Entitäten standardmäßig in der Liste der Dataverse-Tabellen verfügbar. Sie können jedoch zusätzliche Entitäten manuell verfügbar machen:

   1. Wechseln Sie zur Seite "Erweiterte Einstellungen" Ihrer Dataverse-Umgebung.
   2. Klicken Sie oben rechts auf das Filtersymbol, um auf die erweiterte Suche zuzugreifen.
   3. Wählen Sie im Dropdownmenü verfügbare Finanz- und Vorgangsentitäten aus, und klicken Sie dann auf "Ergebnisse".
   4. Wählen Sie die virtuelle Entität aus, die Sie verfügbar machen möchten.
   5. Aktivieren Sie auf der Seite " Entitätsadministrator " die Option "Sichtbar" auf "True", und klicken Sie dann auf " Speichern" und "Schließen".

Nun können Sie die Entität in der Liste der Dataverse-Tabellen mit einem Namen erkennen, der mit mserp_ beginnt.

Konfigurieren von Azure Synapse Link

In diesem Schritt verwenden Sie den Synapse-Link für Dataverse zu Azure Data Lake , um die Tabellen auszuwählen, die Sie aufnehmen möchten. Dieser Dienst ersetzt den Dienst, der früher als Export Data to Azure Data Lake Storage Gen2 bekannt war. Trotz der Benennung verwendet es Azure Synapse Analytics nicht und hängt nicht davon ab. Es handelt sich um einen kontinuierlichen Exportdienst von Dataverse zu ADLS Gen2.

1. Klicken Sie im Power Apps-Portal auf "Analysieren" und dann auf "Mit Azure Synapse verknüpfen".

2. Klicken Sie auf "Neuer Link". Dataverse füllt automatisch deine aktiven Abonnements vom selben Tenant aus. Wählen Sie das entsprechende Abonnement aus der Dropdownliste aus.

3. Aktivieren Sie nicht das Kontrollkästchen "Mit Ihrem Azure Synapse Analytics-Arbeitsbereich verbinden ". Die Daten landen als CSV direkt in deinem ADLS Gen2-Speicherkonto, und für diesen Workflow ist kein Azure Synapse Analytics-Arbeitsbereich erforderlich.

4. Klicken Sie auf der Seite "Synapse Link Creation" auf "Erweitert". Aktivieren Sie dann die Umschaltfläche "Erweiterte Konfigurationseinstellungen anzeigen".

5. Schalten Sie Inkrementelle Updateordnerstruktur ein, und legen Sie den gewünschten Aktualisierungsintervall für Synapse Link fest. Das Minimum beträgt 5 Minuten. Dieses Intervall gilt für alle Tabellen, die in diesem Synapse-Link enthalten sind. (Sie legen einen Zeitplan für Ihre Databricks-Pipeline in einem separaten Schritt fest.)

6. Wähle die Tabellen aus, die du synchronisieren möchtest, und belasse nur die Append- und Partitionierungseinstellungen als Standard.

   • Wenn Sie Daten aus einer Dataverse-nativen App aufnehmen, wählen Sie die relevanten Dataverse-Tabellen direkt aus dem Abschnitt Dataverse aus.
   • Wenn Sie aus F&O erfassen, können Sie entweder direkte Tabellen aus dem Abschnitt D365 Finance & Operations oder virtuelle Entitäten aus dem Abschnitt Dataverse (Präfix mserp_) auswählen. Weitere Informationen zu virtuellen Entitäten finden Sie in Schritt 1.

7. Klicken Sie auf "Speichern". Die anfängliche Synapse Link-Synchronisation beginnt.

   Für F&O-Benutzer kann diese anfängliche Synchronisierung Stunden für große Tabellen mit Hunderten von Gigabyte dauern.

Erstellen Sie eine Entra-ID-Anwendung für die Datenaufnahme

In diesem Schritt erfassen Sie die Entra ID Informationen, die zum Erstellen einer Unity-Katalogverbindung erforderlich sind, die die Aufnahme in Azure Databricks unterstützt.

1. Erfassen Sie die Mandanten-ID Ihres Entra ID-Mandanten (portal.azure.com>>Microsoft Entra ID>>Registerkarte „Übersicht“>>Mandanten-ID, im rechten Bereich aufgeführt).

2. Wenn Sie einen Azure Synapse-Link erstellen, erstellt Azure Synapse einen ADLS-Container zum Synchronisieren der ausgewählten Tabellen. Suchen Sie den NAMEN des ADLS-Containers, indem Sie die Administratorseite des Synapse Link besuchen.

3. Sammeln Sie die Zugangsdaten für den ADLS-Container.

   1. Erstellen Sie eine Microsoft Entra ID-App, wenn Sie noch keines haben.
   2. Sammeln Sie den geheimen Clientschlüssel.
   3. Erfassen Sie die App-ID (portal.azure.com>>Microsoft Entra ID>>Verwalten von>>App-Registrierungen).

4. Gewähren Sie der Entra ID-App Zugriff auf den ADLS-Container, sofern noch nicht geschehen.

   Hinweis

   Stellen Sie sicher, dass Ihre Entra-ID-Anwendung Zugriff auf die ADLS-Container hat, die jedem Synapse Link-Profil zugeordnet sind. Wenn Sie Daten aus mehreren Umgebungen oder Anwendungen aufnehmen, vergewissern Sie sich, dass die Anwendung Rollenzuweisungen für alle relevanten Container hat.

   1. Wechseln Sie zu Azure Storage-Konten , und wählen Sie Ihr Container- oder Speicherkonto aus. (Azure Databricks empfiehlt, Berechtigungen auf Container-Ebene zu verwalten, um das Prinzip der geringsten Rechte einzuhalten.)
   2. Klicken Sie auf Access Control (IAM), und fügen Sie dann die Rollenzuweisung hinzu.
   3. Wählen Sie die Rolle Storage Blob Data Contributor → für Lese-, Schreib- und Löschzugriff aus. Wenn Ihre Organisation dies nicht zulässt, wenden Sie sich an Ihr Azure Databricks Kontoteam.
   4. Klicken Sie auf "Weiter", und wählen Sie dann "Mitglieder auswählen" aus.
   5. Wählen Sie "Benutzer", "Gruppe" oder "Dienstprinzipal" aus, und suchen Sie dann nach Ihrer App-Registrierung. (Wenn die App nicht im Suchergebnis vorhanden ist, können Sie die Objekt-ID explizit in die Suchleiste eingeben und dann die EINGABETASTE drücken.
   6. Klicken Sie auf " Überprüfen" + "Zuweisen".
   7. Um zu bestätigen, dass die Berechtigungen ordnungsgemäß konfiguriert sind, können Sie die Access Control Ihres Containers überprüfen.

Erstellen einer Dynamics 365-Pipeline

Du kannst die Pipeline in der Benutzeroberfläche oder über die API erstellen. Der UI-Assistent verwaltet die Verbindung und die Pipeline gemeinsam, während der API-Pfad sie in zwei separaten Schritten erstellt.

Verwende die Benutzeroberfläche

Der Assistent fordert Sie nach den Entra ID-Anwendungszugangsdaten und Speicherdetails, die Sie in den vorherigen Schritten gesammelt haben, und erstellt dann gemeinsam die Verbindung und die Pipeline.

1. Klicken Sie im linken Menü auf "Neu", und fügen Sie dann Daten hinzu, oder laden Sie sie hoch.
2. Klicken Sie auf der Seite "Daten hinzufügen " auf die Dynamics 365-Kachel .
3. Befolgen Sie von dort aus die Anweisungen im Assistenten.

Verwenden der API

Stelle zuerst die Verbindung her, dann die Pipeline, die sie nutzt. Du brauchst den Verbindungsnamen aus dem ersten Schritt, um die Pipeline im zweiten Schritt zu definieren.

Schritt 1: Eine Dynamics 365-Verbindung herstellen

In diesem Schritt erstellen Sie eine Unity-Katalogverbindung, um Ihre Dynamics 365 Anmeldeinformationen sicher zu speichern und mit der Aufnahme in Azure Databricks zu beginnen.

1. Klicken Sie in Ihrem Arbeitsbereich auf das Katalog.
2. Klicken Sie auf Verbinden, und klicken Sie dann auf "Verbindungen".
3. Klicken Sie auf die Schaltfläche " Verbindung erstellen" .
4. Geben Sie einen eindeutigen Verbindungsnamen an, und wählen Sie dann Dynamics 365 als Verbindungstyp aus.
5. Geben Sie den geheimen Clientschlüssel und die Client-ID der Entra ID-App ein, die im vorherigen Schritt erstellt wurde. Ändern Sie den Bereich nicht. Klicke auf Weiter.
6. Geben Sie den Namen des Azure Storage-Kontos, die Mandanten-ID und den ADLS-Containernamen ein, und klicken Sie dann auf "Verbindung erstellen".
7. Notieren Sie sich den Verbindungsnamen.

Schritt 2: Erstellen Sie die Aufnahmepipeline

In diesem Schritt richten Sie die Aufnahmepipeline ein. Jede aufgenommene Tabelle erhält eine entsprechende Streamingtabelle mit demselben Namen im Ziel. Du kannst entweder ein Notizbuch oder die Databricks-CLI verwenden. Beide Ansätze führen API-Aufrufe an einen Databricks-Dienst durch, der die Pipeline erstellt.

Verwenden eines Notizbuchs

Die Vorlage am Ende dieser Seite definiert Hilfsfunktionen zur Erstellung und Verwaltung der Pipeline. Die erste Zelle richtet diese Funktionen ein, und die zweite ist die Stelle, in der du deine eigene Pipeline definiert.

1. Kopieren Sie die Notizbuchvorlage.
2. Führen Sie die erste Zelle des Notizbuchs aus, ohne sie zu ändern.
3. Ändern Sie die zweite Zelle des Notizbuchs mit den Details zu Ihrer Pipeline (z. B. die Tabelle, die Sie importieren möchten, wo Sie die Daten speichern möchten usw.).
4. Führen Sie die zweite Zelle des Vorlagen-Notebooks aus. Dadurch wird create_pipeline ausgeführt.
5. Sie können ausführen list_pipeline , um die Pipeline-ID und deren Details anzuzeigen.
6. Sie können ausführen edit_pipeline , um die Pipelinedefinition zu bearbeiten.
7. Sie können ausführen delete_pipeline , um die Pipeline zu löschen.

Verwenden der Databricks CLI

So erstellen Sie die Pipeline:

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

So bearbeiten Sie die Pipeline

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

Um die Pipelinedefinition abzurufen:

databricks pipelines get "<your_pipeline_id>"

So löschen Sie die Pipeline

databricks pipelines delete "<your_pipeline_id>"

Um weitere Informationen anzuzeigen, können Sie jederzeit den folgenden Befehl ausführen:

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

Konfigurieren zusätzlicher Features (optional)

Der Connector bietet zusätzliche Features, z. B. SCD Typ 2 für die Verlaufsverfolgung, Auswahl auf Spaltenebene und Entschlüsselung. Siehe allgemeine Muster für verwaltete Aufnahmepipelines.

Notizbuchvorlage

Kopieren Sie beide Zellen in ein Notizbuch in Ihrem Arbeitsbereich. Zelle 1 definiert die Hilfsfunktionen, die die API der Pipelines aufrufen, und Zelle 2 ist der Ort, an dem Sie die Pipeline definieren, die Sie erstellen möchten.

Zelle 1: API-Einrichtung

Kopiere diese Zelle as-is und führe sie ohne Änderungen aus. Sie definiert create_pipeline, list_pipeline, edit_pipeline, , delete_pipeline, und die anderen Helfer, die Zelle 2 anruft.

# 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")

Zelle 2: Pipeline-Definition

Wählen Sie eine der beiden untenstehenden Optionen, je nachdem, wie viel Ihrer Synapse Link-Daten Sie aufnehmen möchten:

• Option A, Schema-Level Spec: Nimmt jede Tabelle ein, die von deinem Azure Synapse Link synchronisiert wird. Azure Databricks empfiehlt nicht mehr als 250 Tabellen pro Pipeline, also wenn dein Synapse Link mehr synchronisiert, teile die Tabellen auf mehrere Pipelines auf.
• Option B, Tabellen-Spezifikation: Nimmt nur die von dir genannten Tabellen ein. Jeder source_table Wert muss mit dem Tabellennamen in der Rubrik Name der Seite Synapse Link verwalten übereinstimmen.

Ersetze die Platzhalterwerte durch deine eigenen, aber lass "channel": "PREVIEW" as-is.

# 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)

# 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)

Beispiel: Streckenhistorie mit SCD Typ 2

Standardmäßig verwendet die API SCD-Typ 1. Dies bedeutet, dass die Daten im Ziel überschrieben werden, wenn sie in der Quelle bearbeitet werden. Wenn Sie historische Daten beibehalten und SCD-Typ 2 verwenden möchten, geben Sie dies in der Konfiguration an. Beispiel:

# 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)

# 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)

Beispiel: Bestimmte Spalten einbeziehen oder ausschließen

Standardmäßig nimmt die API alle Spalten in der ausgewählten Tabelle ein. Sie können jedoch auswählen, dass bestimmte Spalten eingeschlossen oder ausgeschlossen werden sollen. Beispiel:

# 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)