Configuración del origen de datos para la ingestión de Microsoft Dynamics 365

Aprenda a configurar Microsoft Dynamics 365 como origen de datos para la ingesta en Azure Databricks mediante Lakeflow Connect.

Para obtener información sobre cómo el conector accede a los datos de origen, consulte ¿Cómo accede el conector a los datos D365?. Para obtener una lista de las aplicaciones de Dataverse admitidas, consulte ¿Qué aplicaciones de Dynamics 365 se admiten?.

Prerrequisitos

Antes de configurar el origen de datos de Dynamics 365, debe tener:

• Una suscripción de Azure activa con permisos para crear recursos.
• Un entorno de Microsoft Dynamics 365 con acceso de administrador.
• Un entorno de Dataverse asociado a la instancia de Dynamics 365.
• Permisos de administrador del área de trabajo o administrador de metastore en Azure Databricks.
• Permisos para crear y configurar Azure Synapse Link en el entorno de Dataverse.
• Una suscripción a Azure con una cuenta de almacenamiento que no esté ya vinculada a otro perfil de Synapse Link. No puedes añadir tablas de Dataverse a una cuenta de almacenamiento vinculada a otro perfil; debes crear un nuevo perfil en Synapse Link.
• Una cuenta de almacenamiento de ADLS Gen2 (o permisos para crear una).
• Permisos para crear y configurar aplicaciones de Microsoft Entra ID.
• Dataverse API v9.2 o posterior.
• API REST de Azure Storage, versión 2021-08-06.
• Azure Synapse Link para Dataverse versión 1.0 o posterior.

Configuración de entidades virtuales o tablas directas (opcional)

Las entidades virtuales y las tablas directas hacen datos de orígenes que no son de Dataverse (como Dynamics 365 Finance y Operaciones) disponibles en Dataverse sin copiar los datos. Para orígenes que no son de Dataverse, debe configurar entidades virtuales o tablas directas antes de configurar Azure Synapse Link.

Para configurar entidades virtuales:

1. En Power Apps, vaya a la página Entornos y haga clic en Aplicaciones de Dynamics 365.

2. Para vincular entidades de F&O como entidades virtuales en Dataverse, instale la solución Finance and Operations Virtual Entity .

3. Configure la autorización de servicio a servicio (S2S) entre Dataverse y su aplicación de F&O. Esto permite que Dataverse se comunique con la aplicación. Para obtener más información, consulte la documentación de Microsoft Configuración de entidades virtuales de Dataverse.

4. Para cada entidad virtual que quiera ingerir, habilite Seguimiento de cambios en Propiedades avanzadas.

5. De forma predeterminada, la solución entidad virtual de F&O expone algunas entidades virtuales de forma predeterminada en la lista de tablas de Dataverse. Sin embargo, puede exponer entidades adicionales manualmente:

   1. Vaya a la página Configuración avanzada del entorno de Dataverse.
   2. Haga clic en el icono de filtro de la parte superior derecha para acceder a la búsqueda avanzada.
   3. Seleccione Entidades de finanzas y operaciones disponibles en el menú desplegable y, a continuación, haga clic en Resultados.
   4. Seleccione la entidad virtual que desea exponer.
   5. En la página Administrador de entidades, cambie Visible a True y, a continuación, haga clic en Guardar y cerrar.

Ahora puede ver la entidad en la lista de tablas de Dataverse con un nombre que comienza por mserp_.

Configuración de Azure Synapse Link

En este paso, utilizará el enlace de Synapse de Dataverse a Azure Data Lake para seleccionar las tablas que desea importar. Este servicio reemplaza al servicio anteriormente conocido como Exportar datos a Azure Data Lake Storage Gen2. A pesar del nombre, no utiliza ni depende de Azure Synapse Analytics. Es un servicio de exportación continua de Dataverse a ADLS Gen2.

1. En el Portal de Power Apps , haga clic en Analizar y, a continuación, en Vincular a Azure Synapse.

2. Haga clic en Nuevo vínculo. Dataverse rellena automáticamente tus suscripciones activas desde el mismo tenant. Seleccione la suscripción adecuada en la lista desplegable.

3. No active la casilla Conectar al área de trabajo de Azure Synapse Analytics . Los datos aterrizan directamente como CSV en tu cuenta de almacenamiento ADLS Gen2, y este flujo de trabajo no requiere un espacio de trabajo de Azure Synapse Analytics.

4. En la página De creación de vínculos de Synapse, haga clic en Avanzadas. A continuación, active Mostrar opciones de configuración avanzadas.

5. Active la opción Habilitar estructura de carpetas de actualización incremental y establezca el intervalo de actualización deseado para el enlace de Synapse. El mínimo es de 5 minutos. Este intervalo se aplica a todas las tablas incluidas en este enlace Synapse. (Establecerás un horario para tu pipeline de Databricks en un paso independiente).

6. Selecciona las tablas que quieres sincronizar, dejando los ajustes de Añadir solo y Partición como predeterminados.

   • Si se ingiere desde una aplicación nativa de Dataverse, seleccione las tablas de Dataverse pertinentes directamente desde la sección Dataverse .
   • Si la importación se realiza desde F&O, puede seleccionar tablas directas de la sección D365 Finance & Operations o entidades virtuales de la sección Dataverse (prefijo mserp_). Para obtener más información sobre las entidades virtuales, vea Paso 1.

7. Haz clic en Guardar. Comienza la sincronización inicial de Synapse Link.

   Para los usuarios de F&O, esta sincronización inicial puede tardar horas en tablas grandes con cientos de gigabytes.

Creación de una aplicación entra ID para la ingesta

En este paso, recopilarás la información de Entra ID necesaria para crear una conexión de Unity Catalog que admita la ingesta en Azure Databricks.

1. Obtén el ID de inquilino de tu inquilino de Entra ID (portal.azure.com>>Microsoft Entra ID>>pestaña Información general>>ID de inquilino, que aparece en el panel de la derecha).

2. Al crear una instancia de Azure Synapse Link, Azure Synapse crea un contenedor de ADLS para sincronizar las tablas seleccionadas. Busque el nombre del contenedor de ADLS visitando la página de administración del Synapse Link.

3. Recopile las credenciales de acceso para el contenedor de ADLS.

   1. Cree una Aplicación de Microsoft Entra ID, si todavía no tiene una.
   2. Recopile el secreto de cliente.
   3. Recopilar el identificador de aplicación (portal.azure.com>>Microsoft Entra ID>>Administrar>>registros de aplicaciones).

4. Conceda a la aplicación Entra ID acceso al contenedor de ADLS, si aún no lo ha hecho.

   Nota:

   Asegúrese de que la aplicación Entra ID tenga acceso a los contenedores de ADLS asociados a cada perfil de Synapse Link. Si va a ingerir datos de varios entornos o aplicaciones, confirme que la aplicación tiene asignaciones de roles en todos los contenedores pertinentes.

   1. Vaya a Cuentas de Azure Storage y seleccione el contenedor o la cuenta de almacenamiento. (Azure Databricks recomienda el nivel de contenedor para mantener los privilegios mínimos).
   2. Haga clic en Control de acceso (IAM) y, a continuación, en Agregar asignación de roles.
   3. Selecciona la función de acceso Colaborador de datos de Storage Blob → Lectura/Escritura/Eliminación. Si su organización no lo permite, póngase en contacto con el equipo de la cuenta de Azure Databricks.
   4. Haga clic en Siguiente y, a continuación, seleccione Miembros.
   5. Elija Usuario, grupo o entidad de servicio, luego busque su Registro de Aplicaciones. (Si la aplicación no está presente en el resultado de la búsqueda, puede escribir explícitamente su identificador de objeto en la barra de búsqueda y, a continuación, presionar Entrar).
   6. Haga clic en Revisar y asignar.
   7. Para confirmar que los permisos están configurados correctamente, puede comprobar la Access Control del contenedor.

Crea una canalización de Dynamics 365

Puedes crear la canalización en la interfaz o a través de la API. El asistente de la interfaz gestiona la conexión y la tubería juntos, mientras que la ruta de la API los crea como dos pasos separados.

Uso de la interfaz de usuario

El asistente te pide las credenciales de aplicación Entra ID y los detalles de almacenamiento que recogiste en los pasos anteriores, y luego crea la conexión y la pipeline juntos.

1. En el menú de la izquierda, haga clic en Nuevo y, a continuación, en Agregar o cargar datos.
2. En la página Agregar datos , haga clic en el icono de Dynamics 365 .
3. A partir de ahí, sigue las instrucciones del asistente.

Uso de la API

Crea primero la conexión y luego la tubería que la utiliza. Necesitas el nombre de la conexión del primer paso para definir la tubería en el segundo.

Paso 1: Crea una conexión con Dynamics 365

En este paso, creará una conexión de Catálogo de Unity para almacenar de forma segura las credenciales de Dynamics 365 y empezará a realizar la ingesta en Azure Databricks.

1. En tu área de trabajo, haz clic en Catálogo.
2. Haga clic en Conéctese y, a continuación, haga clic en Conexiones.
3. Haga clic en el botón Crear conexión .
4. Proporcione un nombre de conexión único y, a continuación, seleccione Dynamics 365 como tipo de conexión.
5. Escriba el secreto de cliente y el identificador de cliente de la aplicación entra ID creada en el paso anterior. No modifique el ámbito. Haga clic en Siguiente.
6. Escriba el nombre de la cuenta de Azure Storage, el identificador de inquilino y el nombre del contenedor de ADLS y, a continuación, haga clic en Crear conexión.
7. Anote el nombre de la conexión.

Paso 2: Crear la tubería de ingestión

En este paso, configurarás la canalización de ingestión. Cada tabla ingerida obtiene una tabla de streaming correspondiente con el mismo nombre en el destino. Puedes usar tanto un notebook como la CI de Databricks. Ambos enfoques realizan llamadas API a un servicio de Databricks que crea la canalización.

Uso de un cuaderno

La plantilla al final de esta página define funciones auxiliares para crear y gestionar la canalización. La primera celda configura esas funciones, y la segunda es donde defines tu propio pipeline.

1. Copie la plantilla de cuaderno.
2. Ejecute la primera celda del cuaderno sin modificarlo.
3. Modifique la segunda celda del cuaderno con los detalles de la canalización (por ejemplo, la tabla desde la que desea ingerir, donde desea almacenar los datos, etc.).
4. Ejecuta la segunda celda del cuaderno de plantillas; esto ejecuta create_pipeline.
5. Puede ejecutar list_pipeline para mostrar el identificador de canalización y sus detalles.
6. Puede ejecutar edit_pipeline para editar la definición de canalización.
7. Puede ejecutar delete_pipeline para eliminar la canalización.

Uso de la CLI de Databricks

Para crear la canalización:

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

Para editar la canalización:

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

Para obtener la definición de la canalización:

databricks pipelines get "<your_pipeline_id>"

Para eliminar la canalización:

databricks pipelines delete "<your_pipeline_id>"

Para obtener más información, siempre puede ejecutar:

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

Configuración de características adicionales (opcional)

El conector ofrece características adicionales, como SCD tipo 2 para el seguimiento del historial, y la selección y deselección a nivel de columna. Consulte Patrones comunes para canalizaciones de ingesta administradas.

Plantilla de cuaderno

Copia ambas celdas en un cuaderno en tu espacio de trabajo. La celda 1 define las funciones auxiliares que llaman a la API de las pipelines, y la celda 2 es donde defines la pipeline que quieres crear.

Celda 1: Configuración de la API

Copia esta celda as-is y ejecutala sin cambios. Define create_pipeline, list_pipeline, edit_pipeline, delete_pipeline, y los otros ayudantes que llama al celdo 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")

Celda 2: Definición de tubería

Elige una de las dos opciones que aparecen a continuación, dependiendo de cuánto de tus datos de Synapse Link quieras ingerir:

• Opción A, especificación a nivel de esquema: ingiere todas las tablas sincronizadas por tu Azure Synapse Link. Azure Databricks no recomienda más de 250 tablas por pipeline, así que si tu Synapse Link sincroniza más, divide las tablas en varias pipelines.
• Opción B, especialización a nivel de tabla: solo ingiere las tablas que menciones. Cada source_table valor debe coincidir con el nombre de la tabla en la columna de Nombre de la página de Synapse Link Manage.

Sustituye los valores provisionales por los tuyos, pero deja "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)

Ejemplo: Historial de pista con SCD tipo 2

De forma predeterminada, la API usa el tipo 1 de SCD. Esto significa que sobrescribe los datos en el destino si se edita en el origen. Si prefiere conservar los datos históricos y usar el tipo SCD 2, especifique eso en la configuración. Por ejemplo:

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

Ejemplo: Incluir o excluir columnas específicas

De forma predeterminada, la API ingiere todas las columnas de la tabla seleccionada. Sin embargo, puede elegir incluir o excluir columnas específicas. Por ejemplo:

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