Configurare l'origine dati per l'inserimento in Microsoft Dynamics 365

Informazioni su come configurare Microsoft Dynamics 365 come origine dati per l'inserimento in Azure Databricks usando Lakeflow Connect.

Per informazioni su come il connettore accede ai dati di origine, vedere Come il connettore accede ai dati D365?. Per un elenco delle applicazioni Dataverse supportate, vedere Quali applicazioni Dynamics 365 sono supportate?.

Prerequisiti

Prima di configurare l'origine dati Dynamics 365, è necessario disporre di:

• Una sottoscrizione di Azure attiva con autorizzazioni per creare risorse.
• Un ambiente Microsoft Dynamics 365 con accesso amministratore.
• Un ambiente Dataverse associato all'istanza di Dynamics 365.
• Autorizzazioni di amministratore dell'area di lavoro o di amministratore del metastore in Azure Databricks.
• Autorizzazioni per creare e configurare collegamento ad Azure Synapse nell'ambiente Dataverse.
• Un abbonamento Azure con un account di archiviazione che non è già collegato a un altro profilo Collegamento a Synapse. Non puoi aggiungere tabelle Dataverse a un account di archiviazione collegato a un profilo diverso; devi creare un nuovo profilo Collegamento a Synapse.
• Un account di archiviazione ADLS Gen2 (o autorizzazioni per crearne uno).
• Autorizzazioni per creare e configurare applicazioni Microsoft Entra ID.
• API Dataverse v9.2 o versione successiva.
• API REST di Archiviazione di Azure versione 2021-08-06.
• Collegamento ad Azure Synapse per Dataverse versione 1.0 o successiva.

Configurare entità virtuali o tabelle dirette (facoltativo)

Le entità virtuali e le tabelle dirette rendono i dati provenienti da origini non dataverse (ad esempio Dynamics 365 Field Service e Operazioni) disponibili in Dataverse senza copiare i dati. Per le origini non Dataverse, è necessario configurare entità virtuali o tabelle dirette prima di configurare Collegamento ad Azure Synapse.

Per configurare le entità virtuali:

1. In Power Apps passare alla pagina Ambienti e quindi fare clic su App Dynamics 365.

2. Per collegare le entità F&O come entità virtuali in Dataverse, installare la soluzione Entità Virtuale per Finanza e Operazioni.

3. Configurare l'autorizzazione da servizio a servizio (S2S) tra Dataverse e l'applicazione F&O. In questo modo, Dataverse comunica con l'applicazione. Per informazioni dettagliate, vedere la documentazione Microsoft Configurare le entità virtuali di Dataverse.

4. Per ogni entità virtuale che si desidera inserire, abilitare Rileva modifiche in Proprietà avanzate.

5. Per impostazione predefinita, la soluzione F&O Virtual Entity espone alcune entità virtuali per impostazione predefinita nell'elenco di tabelle dataverse. Tuttavia, è possibile esporre manualmente entità aggiuntive:

   1. Passare alla pagina Impostazioni avanzate dell'ambiente Dataverse.
   2. Fare clic sull'icona del filtro in alto a destra per accedere alla ricerca avanzata.
   3. Selezionare Available Finance and Operation Entities (Entità finanza ed operazione disponibili ) dal menu a discesa e quindi fare clic su Risultati.
   4. Selezionare l'entità virtuale da esporre.
   5. Nella pagina Entity Admin (Amministratore entità), impostare Visible (Visibile) su True e quindi fare clic su Save and Close (Salva e chiudi).

È ora possibile visualizzare l'entità nell'elenco delle tabelle dataverse con un nome che inizia con mserp_.

Configurare Collegamento ad Azure Synapse

In questo passaggio si utilizzerà il Collegamento a Synapse per Dataverse ad Azure Data Lake per scegliere le tabelle da acquisire. Questo servizio sostituisce il servizio precedentemente noto come Export data to Azure Data Lake Storage Gen2. Nonostante il nome, non utilizza né dipende da Azure Synapse Analytics. È un servizio di esportazione continua da Dataverse a ADLS Gen2.

1. Nel portale di Power Apps fare clic su Analizza e quindi su Collega ad Azure Synapse.

2. Fare clic su Nuovo collegamento. Dataverse popola automaticamente i tuoi abbonamenti attivi dallo stesso tenant. Selezionare la sottoscrizione appropriata nell'elenco a discesa.

3. Non selezionare la casella di controllo Connetti all'area di lavoro di Azure Synapse Analytics . I dati arrivano direttamente come CSV nel tuo account ADLS Gen2, e questo workflow non richiede uno spazio di lavoro Azure Synapse Analytics.

4. Nella pagina di creazione del collegamento Synapse, fare clic su Avanzate. Quindi, attiva/disattiva Mostra impostazioni di configurazione avanzate.

5. Attiva o disattiva la struttura delle cartelle di aggiornamento incrementale e imposta l'intervallo di aggiornamento desiderato per Collegamento a Synapse. Il minimo è 5 minuti. Questo intervallo si applica a tutte le tabelle incluse in questo collegamento a Synapse. Verrà impostata una pianificazione per la pipeline di Databricks in un passaggio separato.

6. Seleziona le tabelle che vuoi sincronizzare, lasciando le impostazioni Solo Append e Partition come impostazioni predefinite.

   • Se si inserisce da un'app nativa di Dataverse, selezionare direttamente le tabelle pertinenti dalla sezione Dataverse.
   • Se si deve ingestire da F&O, è possibile selezionare tabelle dirette dalla sezione D365 Finance & Operations o entità virtuali direttamente dalla sezione Dataverse (prefisso mserp_). Per altre informazioni sulle entità virtuali, vedere Passaggio 1.

7. Fare clic su Salva. Inizia la sincronizzazione iniziale di Collegamento a Synapse.

   Per gli utenti F&O, questa sincronizzazione iniziale può richiedere ore per tabelle di grandi dimensioni con centinaia di gigabyte.

Creare un'applicazione Entra ID per l'inserimento

In questo passaggio si raccoglieranno le informazioni di Entra ID necessarie per creare una connessione al catalogo Unity che supporta l'inserimento in Azure Databricks.

1. Raccogliere l'ID tenant di Entra ID (portal.azure.com>>ID tenant di Microsoft Entra>>scheda Panoramica>>, elencato nel pannello sulla destra).

2. Quando si crea un collegamento ad Azure Synapse, Azure Synapse crea un contenitore ADLS per sincronizzare le tabelle selezionate. Individuare il nome del contenitore ADLS visitando la pagina di amministrazione di Collegamento a Synapse.

3. Raccogliere le credenziali di accesso per il contenitore ADLS.

   1. Crea un'app Microsoft Entra ID, se non ne hai già una.
   2. Raccogliere il segreto del client.
   3. Raccogliere l'ID app (portal.azure.com>>Microsoft Entra ID>>Gestione>>Registrazioni app).

4. Concedere all'app Entra ID l'accesso al contenitore ADLS, se non è già stato fatto.

   Annotazioni

   Assicurarsi che l'applicazione Entra ID abbia accesso ai contenitori ADLS associati a ogni profilo collegamento a Synapse. Se si inseriscono dati da più ambienti o applicazioni, verificare che l'applicazione disponga di assegnazioni di ruolo in tutti i contenitori pertinenti.

   1. Passare ad Account di archiviazione di Azure e selezionare il contenitore o l'account di archiviazione. (Azure Databricks consiglia il livello del contenitore per mantenere i privilegi minimi.
   2. Fare clic su Controllo di accesso (IAM) e quindi su Aggiungi assegnazione di ruolo.
   3. Selezionare il ruolo di accesso in lettura/scrittura/eliminazione Storage Blob Data Contributor. Se la tua organizzazione non lo consente, contatta il team dell’account di Azure Databricks.
   4. Fare clic su Avanti, quindi selezionare membri.
   5. Scegliere Utente, gruppo o entità servizio, quindi cercare la registrazione dell'app. Se l'app non è presente nel risultato della ricerca, è possibile immettere in modo esplicito il relativo ID oggetto nella barra di ricerca, quindi premere INVIO.
   6. Fare clic su Rivedi e assegna.
   7. Per verificare che le autorizzazioni siano configurate correttamente, è possibile controllare il Controllo di accesso del contenitore.

Creare una pipeline di Dynamics 365

Puoi creare la pipeline nell'interfaccia utente o tramite l'API. Il wizard UI gestisce insieme la connessione e la pipeline, mentre il percorso API li crea in due passaggi separati.

Usare l'interfaccia utente

Il wizard ti chiede le credenziali dell'applicazione Entra ID e i dati di archiviazione raccolti nei passaggi precedenti, poi crea insieme la connessione e la pipeline.

1. Nel menu a sinistra fare clic su Nuovo, quindi su Aggiungi o carica dati.
2. Nella pagina Aggiungi dati fare clic sul riquadro Dynamics 365 .
3. Segui le istruzioni nella procedura guidata da lì.

Usa l'API

Crea prima la connessione, poi la pipeline che la utilizza. Serve il nome della connessione dal primo passaggio per definire la pipeline nel secondo.

Passo 1: Crea una connessione Dynamics 365

In questo passaggio si creerà una connessione del catalogo Unity per archiviare in modo sicuro le credenziali di Dynamics 365 e iniziare l'inserimento in Azure Databricks.

1. Nell'area di lavoro fare clic Catalogo.
2. Fare clic sull'icona , quindi su Connetti, e infine su Connessioni.
3. Fare clic sul pulsante Crea connessione .
4. Specificare un nome di connessione univoco, quindi selezionare Dynamics 365 come tipo di connessione.
5. Immettere il segreto client e l'ID client dell'app Entra ID creata nel passaggio precedente. Non modificare l'ambito. Fare clic su Avanti.
6. Immettere il nome dell'account di archiviazione di Azure, l'ID tenant e il nome del contenitore ADLS, quindi fare clic su Crea connessione.
7. Prendere nota del nome della connessione.

Passo 2: Crea la pipeline di ingestione

In questo passaggio si configurerà la pipeline di inserimento. Ogni tabella inserita ottiene una tabella di streaming corrispondente con lo stesso nome nella destinazione. Puoi usare sia un notebook che la CI di Databricks. Entrambi gli approcci effettuano chiamate API a un servizio Databricks che crea la pipeline.

Usare un notebook

Il modello alla fine di questa pagina definisce le funzioni helper per creare e gestire la pipeline. La prima cella imposta queste funzioni, e la seconda è dove definisci la tua pipeline.

1. Copiare il modello Notebook.
2. Eseguire la prima cella del notebook senza modificarla.
3. Modificare la seconda cella del notebook con i dettagli della pipeline, ad esempio la tabella da inserire, in cui archiviare i dati e così via.
4. Esegui la seconda cella del notebook del modello; questo esegue create_pipeline.
5. È possibile eseguire list_pipeline per visualizzare l'ID della pipeline e i relativi dettagli.
6. È possibile eseguire edit_pipeline per modificare la definizione della pipeline.
7. È possibile eseguire delete_pipeline per eliminare la pipeline.

Usare l'interfaccia della riga di comando di Databricks

Per creare la pipeline:

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

Per modificare la pipeline:

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

Per ottenere la definizione della pipeline:

databricks pipelines get "<your_pipeline_id>"

Per eliminare la pipeline:

databricks pipelines delete "<your_pipeline_id>"

Per altre informazioni, è sempre possibile eseguire:

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

Configurare funzionalità aggiuntive (facoltativo)

Il connettore offre funzionalità aggiuntive, ad esempio scD di tipo 2 per il rilevamento della cronologia, la selezione a livello di colonna e la deselezionazione. Vedere Modelli comuni per le pipeline di inserimento gestite.

Modello di notebook

Copia entrambe le celle in un quaderno nel tuo spazio di lavoro. La cella 1 definisce le funzioni helper che chiamano l'API delle pipeline, e la cella 2 è dove definisci la pipeline che vuoi creare.

Cell 1: Configurazione API

Copia questa cella as-is ed eseguila senza modifiche. Definisce create_pipeline, list_pipeline, edit_pipeline, delete_pipeline, e gli altri aiutanti che chiamano il cellulare 2.

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

Cella 2: Definizione della pipeline

Scegli una delle due opzioni qui sotto, a seconda di quanto dei tuoi dati Collegamento a Synapse vuoi assumere:

• Opzione A, specifica a livello di schema: assorbe ogni tabella sincronizzata dal tuo Azure Collegamento a Synapse. Azure Databricks non consiglia più di 250 tabelle per pipeline, quindi se il tuo Collegamento a Synapse sincronizza più di così, suddividi le tabelle su più pipeline.
• Opzione B, specifica a livello di tabella: ingerisce solo le tabelle che nomini. Ogni source_table valore deve corrispondere al nome della tabella nella colonna Nome della pagina Collegamento a Synapse Manage.

Sostituisci i valori provvisori con i tuoi, ma lascia "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)

Esempio: Storia delle tracce con SCD tipo 2

Per impostazione predefinita, l'API usa il tipo scD 1. Ciò significa che sovrascrive i dati nella destinazione se vengono modificati nell'origine. Se si preferisce conservare i dati storici e utilizzare il tipo SCD 2, specificarlo nella configurazione. Per esempio:

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

Esempio: Includere o escludere colonne specifiche

Per impostazione predefinita, l'API inserisce tutte le colonne nella tabella selezionata. È tuttavia possibile scegliere di includere o escludere colonne specifiche. Per esempio:

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