Microsoft Dynamics 365의 데이터 수집을 위한 데이터 원본 구성

이 페이지에서는 Lakeflow Connect를 사용하여 Azure Databricks에 수집하기 위한 데이터 원본으로 Microsoft Dynamics 365를 설정하는 방법을 설명합니다. Dynamics 365 커넥터는 Dataverse용 Azure Synapse Link를 사용하여 Azure Databricks가 수집하는 ADLS(Azure Data Lake Storage) Gen2로 데이터를 내보냅니다.

필수 조건

D365 데이터 원본을 구성하기 전에 다음이 있어야 합니다.

• 리소스를 만들 수 있는 권한이 있는 활성 Azure 구독입니다.
• 관리자 액세스 권한이 있는 Microsoft Dynamics 365 환경.
• D365 인스턴스와 연결된 Dataverse 환경입니다.
• Azure Databricks의 작업 영역 관리자 또는 메타스토어 관리자 권한
• Dataverse 환경에서 Azure Synapse Link를 만들고 구성할 수 있는 권한입니다.
• ADLS Gen2 스토리지 계정(또는 만들 수 있는 권한)입니다.
• Microsoft Entra ID 애플리케이션을 만들고 구성할 수 있는 권한입니다.
• Dataverse API v9.2 이상.
• Azure Storage REST API 버전 2021-08-06.
• Dataverse용 Azure Synapse Link 버전 1.0 이상.

지원되는 Dynamics 365 애플리케이션은 무엇입니까?

Dynamics 365 커넥터는 두 가지 범주의 애플리케이션을 지원합니다.

Dataverse 네이티브 애플리케이션: 다음과 같은 앱은 Dataverse에 데이터를 직접 저장하며 가상 엔터티 또는 직접 테이블이 필요하지 않습니다.

• Dynamics 365 Sales
• Dynamics 365 고객 서비스
• Dynamics 365 Marketing
• Dynamics 365 Field Service (현장 서비스)

비 Dataverse 네이티브 애플리케이션: 다음과 같은 앱은 Dataverse에서 데이터를 노출하기 위해 가상 엔터티 또는 직접 테이블이 필요합니다.

• Dynamics 365 Finance & Operations(F&O)

1단계: 가상 엔터티 또는 직접 테이블 구성(선택 사항)

가상 엔터티 및 직접 테이블은 데이터를 복사하지 않고 Dataverse에서 비 Dataverse 원본(예: D365 Finance & Operations)의 데이터를 사용할 수 있도록 합니다. 비 Dataverse 원본의 경우 Azure Synapse Link를 설정하기 전에 가상 엔터티 또는 직접 테이블을 구성해야 합니다.

가상 엔터티를 구성하려면 다음을 수행합니다.

1. Power Apps에서 환경 페이지로 이동한 다음 Dynamics 365 앱을 클릭합니다.
2. F&O 엔터티를 Dataverse의 가상 엔터티로 연결하려면 Finance and Operations Virtual Entity 솔루션을 설치합니다.
3. Dataverse와 F&O 애플리케이션 간에 S2S(서비스 간) 권한 부여를 설정합니다. 이를 통해 Dataverse는 애플리케이션과 통신할 수 있습니다. 자세한 내용은 Microsoft 설명서를 참조하세요.

• Dataverse 가상 엔터티 구성

1. 수집하려는 모든 가상 엔터티에 대해 고급 속성에서 변경 내용 추적을 사용하도록 설정합니다.
2. 기본적으로 F&O 가상 엔터티 솔루션은 Dataverse 테이블 목록에 기본적으로 일부 가상 엔터티를 노출합니다. 그러나 추가 엔터티를 수동으로 노출할 수 있습니다.
   1. Dataverse 환경의 고급 설정 페이지로 이동합니다.
   2. 오른쪽 위에 있는 필터 아이콘을 클릭하여 고급 검색에 액세스합니다.
   3. 드롭다운 메뉴에서 사용 가능한 재무 및 운영 엔터티를 선택한 다음 결과를 클릭합니다.
   4. 노출하려는 가상 엔터티를 선택합니다.
   5. 엔터티 관리자 페이지에서 표시를 True로 전환한 다음 저장 및 닫기를 클릭합니다.

이제 이름이 로 시작하는 Dataverse 테이블 목록에서 엔터티를 볼 수 있습니다 mserp_.

2단계: Azure Synapse Link 구성

이 단계에서는 Azure Data Lake에 대한 Dataverse용 Synapse Link 를 사용하여 수집하려는 테이블을 선택합니다.

• Power Apps 포털에서 분석을 클릭한 다음 Azure Synapse에 연결합니다.

• 새 링크를 클릭합니다. Dataverse는 동일한 테넌트에서 활성 구독을 자동으로 업데이트합니다. 드롭다운에서 적절한 구독을 선택합니다.

• Azure Synapse Analytics 작업 영역에 연결 확인란을 선택하지 마세요. (커넥터가 현재 Parquet을 지원하지 않으므로 데이터가 CSV에 있어야 합니다.)

  비고

  이 커넥터를 사용하려면 다른 Synapse Link 프로필에 이미 연결된 기존 스토리지 계정에 Dataverse 테이블을 추가할 수 없습니다. 새 Synapse Link 프로필을 만들 수 있도록 연결되지 않은 Azure 구독에 액세스할 수 있어야 합니다.

• Synapse 링크 만들기 페이지에서고급을 클릭합니다. 그런 다음 고급 구성 설정 표시를 토글합니다.

• 증분 업데이트 폴더 구조를 사용하도록 설정/해제하고 원하는 Synapse Link 업데이트 간격을 설정합니다. 최소 5분입니다. 이 간격은 이 Synapse Link에 포함된 모든 테이블에 적용됩니다. (별도의 단계에서 Databricks 파이프라인에 대한 일정을 설정합니다.)

• 동기화하려는 테이블을 선택하여 추가만 하고 파티션 설정을 기본값으로 둡니다.

  • Dataverse에 기본 내장된 앱에서 수집하는 경우 Dataverse 섹션에서 직접 관련 Dataverse 테이블을 선택합니다.
  • F&O에서 수집하는 경우 D365 Finance & Operations 섹션에서 직접 테이블을 선택하거나 Dataverse 섹션(접두사 mserp_)의 가상 엔터티를 선택할 수 있습니다. 가상 엔터티에 대한 자세한 내용은 1단계를 참조하세요.

• 저장을 클릭합니다.

• 변경 내용을 저장한 후 초기 Synapse Link 동기화가 시작됩니다.

  비고

  F&O 사용자의 경우 이 초기 동기화는 수백 기가바이트가 있는 큰 테이블에 몇 시간이 걸릴 수 있습니다. 그러나 엔터티에 대한 초기 동기화가 너무 오래 걸리는 경우 F&O 앱을 통해 테이블에 인덱스 만들기를 통해 속도를 높일 수 있습니다.

  • F&O 환경에서 인덱싱하려는 테이블로 이동합니다.
  • 테이블에 대한 확장을 만듭니다.
  • 테이블 확장 내에서 새 인덱스 정의
  • 인덱스로 포함할 필드를 추가합니다. 이렇게 하면 이러한 필드를 기반으로 데이터베이스 검색 속도를 높일 수 있습니다.
  • 변경 내용을 저장하고 F&O 환경에 배포합니다.

3단계: 데이터 수집용 Entra ID 애플리케이션 만들기

이 단계에서는 Databricks로의 수집을 지원하는 Unity 카탈로그 연결을 만드는 데 필요한 Entra ID 정보를 수집합니다.

• Entra 테넌트 ID를 수집합니다. 이는 오른쪽 패널에 나열된 portal.azure.com>>개요 탭의 >> 항목에서 찾을 수 있습니다.

• Azure Synapse Link를 만들 때 Azure Synapse는 선택한 테이블을 동기화하기 위한 ADLS 컨테이너를 만듭니다. Synapse Link의 관리 페이지를 방문하여 ADLS 컨테이너 이름을 찾습니다.

• ADLS 컨테이너에 대한 액세스 자격 증명을 수집합니다.

  • 아직 없는 경우 Microsoft Entra ID 앱을 만듭니다.
  • 클라이언트 비밀을 수집합니다.
  • 앱 ID(portal.azure.com>>Microsoft Entra ID>>관리>>앱 등록)를 수집합니다.

• 아직 액세스하지 않은 경우 Entra ID 앱에 ADLS 컨테이너에 대한 액세스 권한을 부여합니다.

  비고

  Entra ID 애플리케이션이 각 Synapse Link 프로필과 연결된 ADLS 컨테이너에 액세스할 수 있는지 확인합니다. 여러 환경 또는 애플리케이션에서 데이터를 수집하는 경우 애플리케이션에 모든 관련 컨테이너에 대한 역할 할당이 있는지 확인합니다.

  • Azure Storage 계정으로 이동하여 컨테이너 또는 스토리지 계정을 선택합니다. (Databricks는 최소 권한을 유지하려면 컨테이너 수준을 권장합니다.)

  • 액세스 제어(IAM)를 클릭한 다음 역할 할당을 추가합니다.

  • Storage Blob 기여자 역할을 선택하여 읽기/쓰기/삭제 액세스 권한을 부여합니다. (조직에서 이를 허용하지 않는 경우 Databricks 계정 팀에 문의하세요.)

  • 다음을 클릭한 다음 구성원을 선택합니다.

  • 사용자, 그룹 또는 서비스 주체를 선택한 다음, 앱 등록을 검색합니다. (앱이 검색 결과에 없는 경우 검색 창에 개체 ID를 명시적으로 입력한 다음 Enter 키를 누릅니다).

  • 검토 + 할당을 클릭합니다.

  • 사용 권한이 올바르게 구성되었는지 확인하려면 컨테이너의 액세스 제어를 확인할 수 있습니다.

4단계: Dynamics 365 파이프라인 만들기

옵션 A: UI 사용

왼쪽 메뉴에서 새로 만들기, 데이터 추가 또는 업로드를 차례로 클릭합니다.

데이터 추가 페이지에서 Dynamics 365 타일을 클릭합니다.

그 다음으로, 마법사에서 안내하는 지침을 따르십시오.

옵션 B: API 사용

B1. Dynamics 365 연결 만들기 이 단계에서는 D365 자격 증명을 안전하게 저장하고 Databricks로 수집을 시작하는 Unity 카탈로그 연결을 만듭니다.

• 작업 영역에서 카탈로그>>외부 데이터>>연결>>연결 만들기를 클릭합니다.

• 고유한 연결 이름을 입력한 다음, Dynamics 365 를 연결 유형으로 선택합니다.

• 이전 단계에서 만든 Entra ID 앱의 클라이언트 암호 및 클라이언트 ID 를 입력합니다. 범위를 수정하지 마세요. 다음을 클릭합니다.

• Azure Storage 계정 이름, 테넌트 ID 및 ADLS 컨테이너 이름을 입력한 다음 연결 만들기를 클릭합니다.

• 연결 이름을 기록해 둡다.

B2. Dynamics 365 파이프라인 만들기

이 단계에서는 인제스트 파이프라인을 설정합니다. 수집된 각 테이블은 대상에서 이름이 같은 해당 스트리밍 테이블을 가져옵니다.

수집 파이프라인을 만드는 두 가지 옵션이 있습니다.

• 전자 필기장 사용
• Databricks CLI(명령줄 인터페이스) 사용

두 방법 모두 파이프라인을 만드는 Databricks 서비스에 대한 API 호출을 만듭니다.

전자 필기장을 사용하려는 경우:

1. 부록에서 Notebook 템플릿을 복사합니다.
2. Notebook의 첫 번째 셀을 수정하지 않고 실행합니다.
3. 파이프라인의 세부 정보(예: 수집하려는 테이블, 데이터 저장 위치 등)를 사용하여 Notebook의 두 번째 셀을 수정합니다.
4. 템플릿 Notebook의 두 번째 셀을 실행하면 create_pipeline이 실행됩니다.
5. 실행 list_pipeline 하여 파이프라인 ID 및 세부 정보를 표시할 수 있습니다.
6. 실행 edit_pipeline 하여 파이프라인 정의를 편집할 수 있습니다.
7. 실행 delete_pipeline 하여 파이프라인을 삭제할 수 있습니다.

Databricks CLI를 사용하려는 경우:

파이프라인을 생성하려면:

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

파이프라인을 편집하려면 다음을 수행합니다.

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

파이프라인 정의를 얻으려면 다음을 수행합니다.

databricks pipelines get "<your_pipeline_id>"

파이프라인을 삭제하려면 다음을 수행합니다.

databricks pipelines delete "<your_pipeline_id>"

자세한 내용은 언제든 다음 명령어를 실행할 수 있습니다.

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

B3. 파이프라인에서 시작하고, 예약하고, 알림을 설정하세요.

파이프라인을 만든 후 Databricks 작업 영역을 다시 방문합니다.

1. 왼쪽 패널에서 파이프라인을 클릭합니다. 새 파이프라인이 이 목록에 나타날 것입니다.
2. 파이프라인 세부 정보를 보려면 파이프라인의 이름을 클릭합니다.
3. 파이프라인 세부 정보 페이지에서 시작을 클릭하여 파이프라인을 즉시 실행할 수 있습니다. 일정을 클릭하여 파이프라인을 예약할 수도 있습니다. 일정 예약에 대한 자세한 내용은 작업 파이프라인 태스크를 참조하세요.

또한 일정에 알림을 추가하여 파이프라인에 알림을 설정할 수도 있습니다.

1. 파이프라인 세부 정보 페이지에서 일정을 클릭합니다. 그런 다음 일정 중 하나를 선택합니다.
2. 당신은 파이프라인 일정의 중심 역할을 맡은 직무로 이동합니다. 오른쪽 패널에서 작업 알림 아래에 원하는 알림을 설정합니다.

마지막으로 지정된 일정에 대한 알림을 추가할 수 있습니다.

1. 새 일정을 만들려면 일정을 클릭합니다.
2. 해당 일정에 알림을 추가하려면 추가 옵션을 클릭합니다.

5단계: 추가 기능 구성(선택 사항)

커넥터는 기록 추적을 위한 SCD 유형 2, 열 수준 선택 및 선택 취소, CI/CD용 Databricks 자산 번들 등의 추가 기능을 제공합니다. 예제는 부록 을 참조하세요.

부록

노트북 템플릿

셀 1

이 셀은 수정하지 마세요.

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

셀 2

아래 코드에서 미리 보기 채널을 수정하지 마세요.

Azure Synapse Link에서 동기화된 모든 테이블을 수집하려면 스키마 수준 사양을 사용합니다. 그러나 Databricks는 파이프라인당 250개 이상의 테이블을 추가하는 것을 권장하지 않습니다. 특정 테이블만 수집하려는 경우 테이블 수준 사양을 사용합니다.

원본 테이블 이름이 Synapse 링크 관리 페이지의 이름 열에 표시되는 테이블 이름과 일치하는지 확인합니다.

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

예: SCD 형식 2

기본적으로 API는 SCD 형식 1을 사용합니다. 즉, 원본에서 편집된 경우 대상의 데이터를 덮어씁니다. 기록 데이터를 유지하고 SCD 형식 2를 사용하려는 경우 구성에서 지정합니다. 다음은 그 예입니다.

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

예: 열 수준 선택 및 선택 취소

기본적으로 API는 선택한 테이블의 모든 열을 수집합니다. 그러나 특정 열을 포함하거나 제외하도록 선택할 수 있습니다. 다음은 그 예입니다.

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