Configurar a fonte de dados para ingestão do Microsoft Dynamics 365

Saiba como configurar o Microsoft Dynamics 365 como uma fonte de dados para ingestão no Azure Databricks usando o Lakeflow Connect.

Para obter informações sobre como o conector acessa seus dados de origem, veja Como o conector acessa dados D365?. Para obter uma lista de aplicativos do Dataverse com suporte, consulte Quais aplicativos do Dynamics 365 têm suporte?.

Pré-requisitos

Antes de configurar a fonte de dados Dynamics 365, você deve ter:

• Uma assinatura ativa do Azure com permissões para criar recursos.
• Um ambiente do Microsoft Dynamics 365 com acesso de administrador.
• Um ambiente do Dataverse associado à instância de Dynamics 365.
• Permissões do administrador de workspace ou administrador do metastore no Azure Databricks.
• Permissões para criar e configurar o Link do Azure Synapse em seu ambiente do Dataverse.
• Uma assinatura do Azure com uma conta de armazenamento que ainda não está vinculada a outro perfil do Link do Synapse. Você não pode adicionar tabelas do Dataverse a uma conta de armazenamento vinculada a um perfil diferente; você deve criar um novo perfil no Link do Synapse.
• Uma conta de armazenamento do ADLS Gen2 (ou permissões para criar uma).
• Permissões para criar e configurar aplicativos de ID do Microsoft Entra.
• API do Dataverse v9.2 ou posterior.
• API REST do Armazenamento do Azure versão 2021-08-06.
• Link do Azure Synapse para Dataverse versão 1.0 ou posterior.

Configurar entidades virtuais ou tabelas diretas (opcional)

Entidades virtuais e tabelas diretas disponibilizam dados de fontes que não são do Dataverse (como o Dynamics 365 Finance & Operations) no Dataverse sem copiar os dados. Para fontes não Dataverse, você deve configurar entidades virtuais ou tabelas diretas antes de configurar o Link do Azure Synapse.

Para configurar entidades virtuais:

1. No Power Apps, vá para a página Ambientes e clique em aplicativos do Dynamics 365.

2. Para vincular entidades F&O como entidades virtuais no Dataverse, instale a solução Entidade Virtual de Finanças e Operações .

3. Configure a autorização S2S (Serviço para Serviço) entre o Dataverse e seu aplicativo F&O. Isso permite que o Dataverse se comunique com seu aplicativo. Para obter detalhes, consulte a documentação da Microsoft Configurar entidades virtuais do Dataverse.

4. Para cada entidade virtual que você deseja ingerir, habilite Rastrear Alterações em Propriedades avançadas.

5. Por padrão, a solução entidade virtual F&O expõe algumas entidades virtuais por padrão na lista de tabelas do Dataverse. No entanto, você pode expor entidades adicionais manualmente:

   1. Vá para a página Configurações Avançadas do ambiente do Dataverse.
   2. Clique no ícone de filtro no canto superior direito para acessar a pesquisa avançada.
   3. Selecione Finanças Disponíveis e Entidades de Operação no menu suspenso e clique em Resultados.
   4. Selecione a entidade virtual que você deseja expor.
   5. Na página Administrador da Entidade , alterne Visible para True e clique em Salvar e Fechar.

Agora você pode ver a entidade na lista de tabelas do Dataverse com um nome que começa com mserp_.

Configurar o Link do Azure Synapse

Nesta etapa, você usará o Link do Synapse para Dataverse no Azure Data Lake para escolher as tabelas que deseja ingerir. Este serviço substitui o serviço anteriormente conhecido como Exportar dados para Azure Data Lake Storage Gen2. Apesar do nome, ele não usa nem depende do Azure Synapse Analytics. É um serviço de exportação contínua do Dataverse para o ADLS Gen2.

1. No Portal do Power Apps , clique em Analisar e, em seguida, vincule ao Azure Synapse.

2. Clique em Novo Link. O Dataverse preenche automaticamente suas assinaturas ativas a partir do mesmo tenant. Selecione a assinatura apropriada na lista suspensa.

3. Não selecione a caixa de seleção Conectar ao Azure Synapse Analytics Workspace. Os dados caem diretamente como CSV na sua conta de armazenamento ADLS Gen2, e esse fluxo de trabalho não exige um workspace do Azure Synapse Analytics.

4. Na Página de Criação de Link do Synapse, clique em Avançado. Em seguida, alterne a opção Mostrar Configurações Avançadas.

5. Ative ou desative Habilitar Estrutura de Pasta de Atualização Incremental e defina o intervalo de atualização desejado para o Link do Synapse. O mínimo é de 5 minutos. Esse intervalo se aplica a todas as tabelas incluídas neste Link do Synapse. (Você definirá um cronograma para seu pipeline do Databricks em uma etapa separada.)

6. Selecione as tabelas que deseja sincronizar, deixando as configurações de Adicionar apenas e Partição como padrão.

   • Ao realizar a ingestão por meio de um aplicativo nativo do Dataverse, selecione as tabelas do Dataverse relevantes diretamente na seção Dataverse.
   • Ao realizar a ingestão por meio do F&O, você pode selecionar tabelas diretamente na seção D365 Finance & Operations ou entidades virtuais da seção Dataverse (prefixo mserp_). Para obter mais informações sobre entidades virtuais, consulte a Etapa 1.

7. Clique em Salvar. A sincronização inicial do Link do Synapse começa.

   Para usuários de F&O, essa sincronização inicial pode levar horas para tabelas grandes com centenas de gigabytes.

Criar um aplicativo de ID do Entra para ingestão

Nesta etapa, você coletará as informações de Entra ID necessárias para criar uma conexão do Catálogo do Unity que dê suporte à ingestão em Azure Databricks.

1. Colete a ID do locatário do seu locatário do Entra ID (portal.azure.com>>Microsoft Entra ID>>Guia Visão geral>>ID do locatário, listadas no painel à direita).

2. Quando você cria um Link do Azure Synapse, o Azure Synapse cria um contêiner do ADLS para sincronizar as tabelas selecionadas. Localize o nome do contêiner do ADLS visitando a Página de Administração do Link do Synapse.

3. Colete as credenciais de acesso para o contêiner do ADLS.

   1. Crie um Aplicativo de ID do Microsoft Entra, se você ainda não tiver um.
   2. Colete o segredo do cliente.
   3. Colete a ID do aplicativo (portal.azure.com>>Microsoft Entra ID>>Manage>>App Registrations).

4. Conceda ao aplicativo Entra ID acesso ao contêiner do ADLS, caso ainda não tenha feito isso.

   Observação

   Verifique se o aplicativo Entra ID tem acesso aos contêineres do ADLS associados a cada perfil do Link do Synapse. Se você estiver ingerindo dados de vários ambientes ou aplicativos, confirme se o aplicativo tem atribuições de função em todos os contêineres relevantes.

   1. Acesse as Contas de Armazenamento do Azure e selecione sua conta de contêiner ou armazenamento. (Azure Databricks recomenda o nível de contêiner para manter privilégios mínimos.)
   2. Clique em Controle de Acesso (IAM) e, em seguida, adicione a atribuição de função.
   3. Selecione a função Colaborador de dados de Storage Blob → Acesso de leitura/gravação/exclusão. Se sua organização não permitir isso, entre em contato com sua equipe de conta Azure Databricks.
   4. Clique em Avançar e selecione Membros.
   5. Escolha Usuário, grupo ou entidade de serviço, então procure seu Registro de Aplicativo. (Se o aplicativo não estiver presente no resultado da pesquisa, você poderá inserir explicitamente sua ID de objeto na barra de pesquisa e pressionar Enter).
   6. Clique em Revisar + Atribuir.
   7. Para confirmar se as permissões estão configuradas corretamente, você pode verificar o Controle de Acesso do contêiner.

Criar um pipeline do Dynamics 365

Você pode criar o pipeline na interface ou pela API. O assistente de interface gerencia a conexão e o pipeline juntos, enquanto o caminho da API os cria como duas etapas separadas.

Usar a interface do usuário

O assistente solicita as credenciais do aplicativo Entra ID e os detalhes de armazenamento que você reuniu nas etapas anteriores, então cria a conexão e o pipeline juntos.

1. No menu à esquerda, clique em Novo e em Adicionar ou carregar dados.
2. Na página Adicionar dados, clique no bloco Dynamics 365.
3. Siga as instruções do assistente a partir daí.

"Utilize a API"

Crie a conexão primeiro, depois o pipeline que a utiliza. Você precisa do nome da conexão do primeiro passo para definir o pipeline no segundo.

Passo 1: Crie uma conexão Dynamics 365

Nesta etapa, você criará uma conexão do Catálogo do Unity para armazenar com segurança suas credenciais de Dynamics 365 e iniciar a ingestão em Azure Databricks.

1. No seu espaço de trabalho, clique no Catálogo.
2. Clique Conecte-se e clique em Conexões.
3. Clique no botão Criar conexão .
4. Forneça um nome de conexão exclusivo e selecione Dynamics 365 como o tipo de conexão.
5. Insira o segredo do cliente e a ID do cliente do aplicativo Entra ID criado na etapa anterior. Não modifique o escopo. Clique em Próximo.
6. Insira o nome da conta de armazenamento do Azure, a ID do locatário e o nome do contêiner do ADLS e clique em Criar Conexão.
7. Anote o nome da conexão.

Passo 2: Criar o pipeline de ingestão

Nesta etapa, você configurará o canal de ingestão. Cada tabela ingerida obtém uma tabela de streaming correspondente com o mesmo nome no destino. Você pode usar tanto um notebook quanto a linha de comando do Databricks. Ambas as abordagens fazem chamadas à API para um serviço do Databricks que cria o pipeline.

Usar um notebook

O modelo ao final desta página define funções auxiliares para criar e gerenciar o pipeline. A primeira célula configura essas funções, e a segunda é onde você define seu próprio pipeline.

1. Copie o modelo de Bloco de Anotações.
2. Execute a primeira célula do notebook sem modificá-la.
3. Modifique a segunda célula do notebook com os detalhes do pipeline (por exemplo, a tabela da qual você deseja fazer a ingestão, onde deseja armazenar os dados, etc.).
4. Execute a segunda célula do notebook de modelo; isso executa create_pipeline.
5. Você pode executar list_pipeline para mostrar a ID do pipeline e seus detalhes.
6. Você pode executar edit_pipeline para editar a definição de pipeline.
7. Você pode executar delete_pipeline para excluir o pipeline.

Como usar a CLI do Databricks

Para criar o pipeline:

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

Para editar o pipeline:

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

Para obter a definição do pipeline:

databricks pipelines get "<your_pipeline_id>"

Para excluir o pipeline:

databricks pipelines delete "<your_pipeline_id>"

Para obter mais informações, você sempre pode executar:

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

Configurar recursos adicionais (opcional)

O conector oferece recursos adicionais, como SCD tipo 2 para rastreamento de histórico e seleção ou desseleção em nível de coluna. Consulte Padrões comuns para pipelines de ingestão gerenciada.

Modelo de caderno

Copie ambas as células para um caderno no seu espaço de trabalho. A célula 1 define as funções auxiliares que chamam a API do pipeline, e a célula 2 é onde você define o pipeline que deseja criar.

Célula 1: Configuração da API

Copie essa célula as-is e a execute sem alterações. Ele define create_pipeline, list_pipeline, edit_pipeline, delete_pipeline, e os outros auxiliares que o celular 2 chama.

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

Célula 2: Definição do pipeline

Escolha uma das duas opções abaixo, dependendo de quanto dos seus dados do Link do Synapse você deseja ingerir:

• Opção A, especificação em nível de esquema: ingere todas as tabelas sincronizadas pelo seu Azure Link do Synapse. O Azure Databricks não recomenda mais de 250 tabelas por pipeline, então se seu Link do Synapse sincroniza mais do que isso, divida as tabelas em vários pipelines.
• Opção B, especificação em nível de mesa: ingere apenas as tabelas que você nomear. Cada source_table valor deve corresponder ao nome da tabela na coluna Nome da página Link do Synapse Manage.

Substitua os valores provisórios pelos seus, mas deixe "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)

Exemplo: Histórico de pista com SCD tipo 2

Por padrão, a API usa o tipo SCD 1. Isso significa que os dados no destino são substituídos se forem editados na origem. Se você preferir preservar dados históricos e usar o SCD tipo 2, especifique-os na configuração. Por exemplo:

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

Exemplo: Incluir ou excluir colunas específicas

Por padrão, a API ingere todas as colunas na tabela selecionada. No entanto, você pode optar por incluir ou excluir colunas específicas. Por exemplo:

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