Konfiguracja źródła danych do wczytywania danych dla Microsoft Dynamics 365

Dowiedz się, jak skonfigurować usługę Microsoft Dynamics 365 jako źródło danych na potrzeby pozyskiwania danych do usługi Azure Databricks przy użyciu programu Lakeflow Connect.

Aby uzyskać informacje o sposobie uzyskiwania dostępu do danych źródłowych przez łącznik, zobacz Jak łącznik uzyskuje dostęp do danych D365?. Aby zapoznać się z listą obsługiwanych aplikacji usługi Dataverse, zobacz Które aplikacje usługi Dynamics 365 są obsługiwane?.

Wymagania wstępne

Przed skonfigurowaniem źródła danych Dynamics 365 musisz mieć następujące elementy:

• Aktywna subskrypcja platformy Azure z uprawnieniami do tworzenia zasobów.
• Środowisko usługi Microsoft Dynamics 365 z dostępem administratora.
• Środowisko Dataverse skojarzone z wystąpieniem usługi Dynamics 365.
• Uprawnienia administratora obszaru roboczego lub administratora magazynu metadanych w usłudze Azure Databricks.
• Uprawnienia do tworzenia i konfigurowania usługi Azure Synapse Link w środowisku usługi Dataverse.
• Subskrypcja Azure z kontem magazynującym, które nie jest jeszcze powiązane z innym profilem Synapse Link. Nie możesz dodawać tabel Dataverse do konta pamięci masowej powiązanego z innym profilem; musisz założyć nowy profil Synapse Link.
• Konto magazynowe ADLS Gen2 (lub uprawnienia do jego utworzenia).
• Uprawnienia do tworzenia i konfigurowania aplikacji Microsoft Entra ID.
• Interfejs API usługi Dataverse w wersji 9.2 lub nowszej.
• Interfejs API REST usługi Azure Storage w wersji 2021-08-06.
• Usługa Azure Synapse Link dla usługi Dataverse w wersji 1.0 lub nowszej.

Konfigurowanie jednostek wirtualnych lub tabel bezpośrednich (opcjonalnie)

Jednostki wirtualne i tabele bezpośrednie udostępniają w usłudze Dataverse dane ze źródeł innych niż Dataverse (na przykład Dynamics 365 Finance & Operations) bez kopiowania tych danych. W przypadku źródeł innych niż Dataverse należy skonfigurować jednostki wirtualne lub tabele bezpośrednie przed skonfigurowaniem usługi Azure Synapse Link.

Aby skonfigurować jednostki wirtualne:

1. W usłudze Power Apps przejdź do strony Środowiska , a następnie kliknij pozycję Aplikacje usługi Dynamics 365.

2. Aby połączyć jednostki F&O jako jednostki wirtualne w usłudze Dataverse, zainstaluj rozwiązanie Finance and Operations Virtual Entity .

3. Skonfiguruj autoryzację Service To Service (S2S) między Dataverse a aplikacją F&O. Dzięki temu usługa Dataverse komunikuje się z aplikacją. Aby uzyskać szczegółowe informacje, zobacz dokumentację firmy Microsoft Konfigurowanie jednostek wirtualnych usługi Dataverse.

4. Dla każdej jednostki wirtualnej, którą chcesz pozyskać, włącz opcję Śledź zmiany w obszarze Właściwości zaawansowane.

5. Domyślnie rozwiązanie jednostki wirtualnej F&O uwidacznia niektóre jednostki wirtualne domyślnie na liście tabel usługi Dataverse. Można jednak ręcznie uwidocznić dodatkowe jednostki:

   1. Przejdź do strony Ustawienia zaawansowane środowiska Dataverse.
   2. Kliknij ikonę filtru w prawym górnym rogu, aby uzyskać dostęp do wyszukiwania zaawansowanego.
   3. Wybierz pozycję Dostępne jednostki finansowe i operacyjne z menu rozwijanego, a następnie kliknij pozycję Wyniki.
   4. Wybierz jednostkę wirtualną, którą chcesz uwidocznić.
   5. Na stronie Administrator jednostki przełącz pozycję Widoczne na wartość True, a następnie kliknij pozycję Zapisz i zamknij.

Teraz możesz wyświetlić jednostkę na liście tabel Usługi Dataverse z nazwą rozpoczynającą się od mserp_.

Konfigurowanie usługi Azure Synapse Link

W tym kroku użyjesz Synapse Link dla Dataverse do Azure Data Lake, aby wybrać tabele, które chcesz zaimportować. Ta usługa zastępuje usługę wcześniej znaną jako Export data do Azure Data Lake Storage Gen2. Pomimo nazwy, nie korzysta ani nie zależy od Azure Synapse Analytics. To usługa ciągłego eksportu z Dataverse do ADLS Gen2.

1. W portalu usługi Power Apps kliknij pozycję Analizuj, a następnie pozycję Połącz z usługą Azure Synapse.

2. Kliknij pozycję Nowy link. Dataverse automatycznie uzupełnia twoje aktywne subskrypcje z tej samej dzierżawy. Wybierz odpowiednią subskrypcję z listy rozwijanej.

3. Nie zaznaczaj pola wyboru Połącz z obszarem roboczym usługi Azure Synapse Analytics . Dane trafiają jako pliki CSV bezpośrednio do Twojego konta magazynu ADLS Gen2, a ten workflow nie wymaga przestrzeni roboczej Azure Synapse Analytics.

4. Na stronie Tworzenia linku usługi Synapse kliknij pozycję Zaawansowane. Następnie przełącz przełącznik Pokaż zaawansowane ustawienia konfiguracji.

5. Przełącz opcję Włącz strukturę folderów aktualizacji przyrostowej i ustaw żądany interwał aktualizacji usługi Synapse Link. Minimalna wartość to 5 minut. Ten interwał dotyczy wszystkich tabel zawartych w tym usłudze Synapse Link. (W osobnym kroku ustawisz harmonogram potoku usługi Databricks).

6. Wybierz tabele, które chcesz zsynchronizować, pozostawiając ustawienia Tylko dołączanie i Partycja jako domyślne.

   • W przypadku pobierania danych z aplikacji natywnej Dataverse wybierz odpowiednie tabele Dataverse bezpośrednio w sekcji Dataverse.
   • W przypadku pozyskiwania z poziomu F&O możesz wybrać tabele bezpośrednie w sekcji Finanse i operacje D365 lub jednostki wirtualne z sekcji Dataverse (prefiks mserp_). Aby uzyskać więcej informacji na temat jednostek wirtualnych, zobacz Krok 1.

7. Kliknij przycisk Zapisz. Rozpoczyna się początkowa synchronizacja Synapse Link.

   W przypadku użytkowników F&O ta początkowa synchronizacja może potrwać kilka godzin w przypadku dużych tabel z setkami gigabajtów.

Tworzenie aplikacji Entra ID na potrzeby ingestii

W tym kroku zbierzesz informacje o identyfikatorze Entra ID potrzebne do utworzenia połączenia z usługą Unity Catalog, które obsługuje pozyskiwanie danych do usługi Azure Databricks.

1. Zbierz identyfikator dzierżawcy swojej dzierżawy portal.azure.com>>Microsoft Entra ID>>(karta Przegląd)>>Identyfikator dzierżawcy, widoczny w panelu po prawej stronie.

2. Podczas tworzenia usługi Azure Synapse Link usługa Azure Synapse tworzy kontener usługi ADLS do synchronizowania wybranych tabel. Znajdź nazwę kontenera usługi ADLS, odwiedzając stronę administratora Synapse Link.

3. Zbierz poświadczenia dostępu dla kontenera usługi ADLS.

   1. Utwórz aplikację Microsoft Entra ID, jeśli jeszcze jej nie masz.
   2. Zbierz klucz tajny klienta.
   3. Zbierz identyfikator aplikacji (portal.azure.com>>Microsoft Entra ID>>Zarządzanie>>rejestracjami aplikacji).

4. Udziel aplikacji Entra ID dostęp do kontenera usługi ADLS, jeśli jeszcze tego nie zrobiono.

   Uwaga / Notatka

   Upewnij się, że aplikacja Entra ID ma dostęp do kontenerów usługi ADLS skojarzonych z każdym profilem usługi Synapse Link. Jeśli pozyskujesz dane z wielu środowisk lub aplikacji, upewnij się, że aplikacja ma przypisania ról we wszystkich odpowiednich kontenerach.

   1. Przejdź do Konta usługi Azure Storage i wybierz swój kontener lub konto magazynowania. (Azure Databricks zaleca poziom kontenera, aby zachować najmniejsze uprawnienia).
   2. Kliknij pozycję Kontrola dostępu (Zarządzanie dostępem i tożsamościami), a następnie dodaj przypisanie roli.
   3. Wybierz rolę Współautor danych obiektów blob usługi Storage → rola dostępu z uprawnieniami odczytu/zapisu/usuwania. Jeśli twoja organizacja tego nie zezwala, skontaktuj się z zespołem ds. kont Azure Databricks.
   4. Kliknij przycisk Dalej, a następnie wybierz pozycję Członkowie.
   5. Wybierz pozycję Użytkownik, grupa lub jednostka usługi, a następnie wyszukaj rejestrację aplikacji. (Jeśli aplikacja nie jest obecna w wynikach wyszukiwania, możesz jawnie wprowadzić jego identyfikator obiektu na pasku wyszukiwania, a następnie nacisnąć Enter).
   6. Kliknij Przejrzyj i przypisz.
   7. Aby upewnić się, że uprawnienia są poprawnie skonfigurowane, możesz sprawdzić Access Control kontenera.

Tworzenie potoku usługi Dynamics 365

Możesz tworzyć potok w interfejsie użytkownika lub przez API. Kreator UI obsługuje połączenie i potok danych razem, podczas gdy ścieżka API tworzy je jako dwa oddzielne kroki.

Korzystanie z interfejsu użytkownika

Kreator prosi cię o podanie danych uwierzytelniających aplikacji Entra ID oraz danych dotyczących przechowywania, które zebrałeś w poprzednich krokach, a następnie tworzy połączenie i potok razem.

1. W menu po lewej stronie kliknij pozycję Nowy, a następnie dodaj lub przekaż dane.
2. Na stronie Dodawanie danych kliknij kafelek Dynamics 365 .
3. Następnie postępuj zgodnie z instrukcjami w kreatorze.

Użyj interfejsu API

Najpierw utwórz połączenie, potem potok przetwarzania, który go używa. Potrzebujesz nazwy połączenia z pierwszego kroku, aby zdefiniować pipeline w drugim.

Krok 1: Stwórz połączenie Dynamics 365

W tym kroku utworzysz połączenie Unity Catalog, aby bezpiecznie przechowywać poświadczenia Dynamics 365 i rozpocząć ładowanie danych do Azure Databricks.

1. W obszarze roboczym kliknij Katalog.
2. Kliknij Połącz, a następnie kliknij Połączenia.
3. Kliknij przycisk Utwórz połączenie .
4. Podaj unikatową nazwę połączenia, a następnie wybierz pozycję Dynamics 365 jako typ połączenia.
5. Wprowadź klucz tajny klienta i identyfikator klienta aplikacji Entra ID utworzonej w poprzednim kroku. Nie modyfikuj zakresu. Kliknij przycisk Dalej.
6. Wprowadź nazwę konta usługi Azure Storage, identyfikator dzierżawy i nazwę kontenera usługi ADLS, a następnie kliknij pozycję Utwórz połączenie.
7. Zanotuj nazwę połączenia.

Krok 2: Utwórz potok pozyskiwania danych

W tym kroku skonfigurujesz potok przesyłania danych. Każda załadowana tabela otrzymuje odpowiadającą jej tabelę przesyłania strumieniowego o tej samej nazwie w lokalizacji docelowej. Możesz użyć notatnika lub CLI Databricks. Oba podejścia wykonują wywołania interfejsu API do usługi Databricks, która tworzy potok.

Użyj notesu

Szablon na końcu tej strony definiuje funkcje pomocnicze do tworzenia i zarządzania pipeline. Pierwsza komórka ustawia te funkcje, a druga to miejsce, gdzie definiujesz własny pipeline.

1. Skopiuj szablon notesu.
2. Uruchom pierwszą komórkę notatnika bez modyfikowania.
3. Zmodyfikuj drugą komórkę notesu z informacjami o potoku (na przykład tabelą, z której chcesz pobierać dane, itp.).
4. Uruchom drugą komórkę notesu szablonu; spowoduje to uruchomienie create_pipeline polecenia.
5. Możesz uruchomić polecenie list_pipeline , aby wyświetlić identyfikator potoku i jego szczegóły.
6. Możesz uruchomić polecenie edit_pipeline, aby edytować definicję pipeline'u.
7. Możesz uruchomić polecenie delete_pipeline, aby usunąć pipeline.

Korzystanie z interfejsu wiersza polecenia usługi Databricks

Aby utworzyć potok danych:

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

Aby edytować przepływ:

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

Aby otrzymać definicję przepływu danych:

databricks pipelines get "<your_pipeline_id>"

Aby usunąć rurociąg:

databricks pipelines delete "<your_pipeline_id>"

Aby uzyskać więcej informacji, zawsze można uruchomić:

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

Konfigurowanie dodatkowych funkcji (opcjonalnie)

Łącznik oferuje dodatkowe funkcje, takie jak SCD type 2 do śledzenia historii, wybór i usuwanie wyboru na poziomie kolumn. Zobacz Typowe wzorce dla modeli zarządzanych potoków danych.

Szablon notesu

Skopiuj obie komórki do notatnika w swoim obszarze roboczym. Komórka 1 definiuje funkcje pomocnicze wywołujące API pipelines, a komórka 2 to miejsce, gdzie definiujesz pipeline, który chcesz utworzyć.

Sekcja 1: Konfiguracja API

Skopiuj tę komórkę as-is i uruchom ją bez zmian. Definiuje create_pipeline, list_pipeline, edit_pipeline, delete_pipeline oraz inne pomocniki wywoływane przez komórkę 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")

Komórka 2: Definicja pipeline'u

Wybierz jedną z dwóch poniższych opcji, w zależności od tego, ile danych Synapse Link chcesz zaimportować:

• Opcja A, specyfikacja na poziomie schematu: importuje każdą tabelę zsynchronizowaną przez Twój Azure Synapse Link. Azure Databricks nie zaleca używania więcej niż 250 tabel w jednym potoku, więc jeśli Twój Synapse Link synchronizuje więcej niż 250 tabel, podziel tabele na kilka potoków.
• Opcja B, specyfikacja na poziomie tabeli: pobiera tylko te tabele, które wymienisz. Każda source_table wartość musi odpowiadać nazwie tabeli w kolumnie Nazwa na stronie Zarządzaj Synapse Link.

Zastąp wartości zastępcze własnymi.

# 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>"
          }
        }
      ]
 }
}
"""

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>"
          }
        }
      ]
 }
}
"""

create_pipeline(pipeline_spec)

Przykład: Śledzenie historii z SCD typu 2

Domyślnie interfejs API używa typu SCD 1. Oznacza to, że zastępuje dane w miejscu docelowym, gdy zostaną zmodyfikowane w źródle. Jeśli wolisz zachować dane historyczne i użyć typu SCD 2, określ je w konfiguracji. Przykład:

# 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"
            }
          }
        }
      ]
 }
}
"""

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"
            }
          }
        }
      ]
 }
}
"""

create_pipeline(pipeline_spec)

Przykład: Uwzględnij lub wyklucz konkretne kolumny

Domyślnie interfejs API pozyskuje wszystkie kolumny w wybranej tabeli. Można jednak dołączać lub wykluczać określone kolumny. Przykład:

# 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>"]
            }
          }
        }
      ]
 }
}
"""

create_pipeline(pipeline_spec)