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

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

Para informações sobre como o conector acede aos seus dados de origem, veja Como é que o conector acede aos dados D365?. Para uma lista de aplicações Dataverse suportadas, veja Que aplicações Dynamics 365 são suportadas?.

Pré-requisitos

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

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

Configurar entidades virtuais ou tabelas diretas (opcional)

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

Para configurar entidades virtuais:

1. No Power Apps, vai à página de Ambientes e depois clica em Apps do Dynamics 365.

2. Para ligar entidades F&O como entidades virtuais no Dataverse, instale a solução Finance and Operations Virtual Entity .

3. Configura a autorização Service To Service (S2S) entre o Dataverse e a tua aplicação F&O. Isto permite ao Dataverse comunicar com a sua aplicação. Para mais detalhes, consulte a documentação da Microsoft Configurar entidades virtuais Dataverse.

4. Para cada entidade virtual que pretende ingerir, ative o Rastreio de Alterações em Propriedades Avançadas.

5. Por defeito, a solução F&O Virtual Entity expõe algumas entidades virtuais por defeito na lista de tabelas Dataverse. No entanto, pode expor entidades adicionais manualmente:

   1. Vai à página de Definições Avançadas do teu ambiente Dataverse.
   2. Clique no ícone do filtro no canto superior direito para aceder à pesquisa avançada.
   3. Selecione Entidades Financeiras e Operacionais Disponíveis no menu suspenso e depois clique em Resultados.
   4. Seleciona a entidade virtual que queres expor.
   5. Na página de Administrador da Entidade , alterne Visível para Verdadeiro, depois clique em Guardar e Fechar.

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

Configurar Azure Synapse Link

Neste passo, irá usar o Synapse Link do Dataverse para o Azure Data Lake para escolher as tabelas que pretende ingerir. Este serviço substitui o serviço anteriormente conhecido como Exportar dados para Azure Data Lake Storage Gen2. Apesar da nomeação, não utiliza nem depende do Azure Synapse Analytics. É um serviço de exportação contínua do Dataverse para o ADLS Gen2.

1. No Portal Power Apps , clique em Analisar, depois Ligue para o Azure Synapse.

2. Clique em Novo Link. O Dataverse preenche automaticamente as suas subscrições ativas a partir do mesmo inquilino. Selecione a subscrição apropriada no menu suspenso.

3. Não selecione a caixa de seleção Ligar ao seu Azure Synapse Analytics Workspace. Os dados aparecem diretamente como CSV na tua conta de armazenamento ADLS Gen2, e este fluxo de trabalho não requer um espaço de trabalho Azure Synapse Analytics.

4. Na página de Criação de Links Synapse, clique em Avançado. Depois, desativa Mostrar Definições Avançadas de Configuração.

5. Alternar Ativar Estrutura de Pastas de Atualização Incremental e definir o intervalo desejado de atualização do Synapse Link. O mínimo é de 5 minutos. Este intervalo aplica-se a todas as tabelas incluídas neste Synapse Link. (Vai definir um horário para o seu pipeline Databricks numa etapa separada.)

6. Seleciona as tabelas que queres sincronizar, deixando as definições de Anexar apenas e Partição como padrão.

   • Se estiver a realizar ingestão a partir de uma aplicação nativa do Dataverse, selecione diretamente da secção Dataverse as tabelas relevantes do Dataverse.
   • Se estiver a importar do F&O, pode selecionar tabelas diretas da secção Finanças & Operações do D365 ou entidades virtuais da secção Dataverse (prefixo mserp_). Para mais informações sobre entidades virtuais, consulte o Passo 1.

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

   Para utilizadores de F&O, esta sincronização inicial pode demorar horas em grandes tabelas com centenas de gigabytes.

Crie uma aplicação Entra ID para ingestão

Neste passo, irá recolher a informação do Entra ID necessária para criar uma ligação ao Unity Catalog que suporta a ingestão no Azure Databricks.

1. Recolha o ID do inquilino do seu inquilino do Entra ID (portal.azure.com>>Microsoft Entra ID>>separador de visão geral>>ID do inquilino, listado no painel direito).

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

3. Recolha as credenciais de acesso para o contentor ADLS.

   1. Crie uma aplicação Microsoft Entra ID, se ainda não tiver uma.
   2. Recolhe o segredo do cliente.
   3. Recolha o ID da App (portal.azure.com>>Microsoft Entra ID>>Gerir>>Registos de Aplicações).

4. Conceda à aplicação Entra ID acesso ao contentor ADLS, caso ainda não o tenha feito.

   Observação

   Certifique-se de que a sua aplicação Entra ID tem acesso aos contentores ADLS associados a cada perfil Synapse Link. Se estiver a ingerir dados de múltiplos ambientes ou aplicações, confirme que a aplicação tem atribuições de funções em todos os contentores relevantes.

   1. Vá a Contas de Armazenamento do Azure e selecione a sua conta de contentor ou de armazenamento. (O Azure Databricks recomenda que o nível do contentor mantenha os privilégios mínimos.)
   2. Clica em Controlo de Acesso (IAM) e depois Adicionar atribuição de funções.
   3. Selecione o Contribuidor de Dados do Blob de Armazenamento → função de acesso Ler/Escrever/Eliminar. Se a sua organização não permitir isto, contacte a sua equipa de contas Azure Databricks.
   4. Clique em Próximo, depois selecione Membros.
   5. Escolha Utilizador, grupo ou principal de serviço, depois Pesquise pelo Registo da Aplicação. (Se a aplicação não estiver presente no resultado da pesquisa, pode inserir explicitamente o ID do objeto na barra de pesquisa e depois pressionar Enter).
   6. Clique em Rever + Atribuir.
   7. Para confirmar que as permissões estão configuradas corretamente, pode verificar o Controlo de Acesso do seu contentor.

Criar um fluxo de trabalho Dynamics 365

Pode criar o pipeline na interface ou através da API. O assistente de interface trata da ligação e do pipeline em conjunto, enquanto o caminho da API cria-os como dois passos separados.

Usar a interface do usuário

O assistente pede-te as credenciais da aplicação Entra ID e os detalhes de armazenamento que recolheste nos passos anteriores, depois cria a ligação e o pipeline em conjunto.

1. No menu esquerdo, clique em Novo, depois Adicionar ou carregar dados.
2. Na página Adicionar dados, clique no azulejo Dynamics 365.
3. Segue as instruções do feiticeiro a partir daí.

Utilizar a API

Crie primeiro a ligação, depois o pipeline que a utiliza. Precisas do nome da ligação do primeiro passo para definir o pipeline no segundo.

Passo 1: Criar uma ligação Dynamics 365

Neste passo, irá criar uma ligação ao Unity Catalog para armazenar de forma segura as suas credenciais do Dynamics 365 e iniciar a ingestão no Azure Databricks.

1. No seu espaço de trabalho, clique Catálogo.
2. Clica Liga, depois clica em Ligações.
3. Clica no botão Criar ligação .
4. Forneça um nome de Ligação único e depois selecione Dynamics 365 como tipo de Ligação.
5. Introduza o segredo do cliente e o ID do cliente da aplicação Entra ID criados na etapa anterior. Não modifiquem o telescópio. Clique em Next.
6. Introduza o Nome da Conta de Armazenamento Azure, o ID do Tenant e o nome do contentor ADLS, depois clique em Criar Ligação.
7. Regista o nome da ligação.

Passo 2: Criar o pipeline de ingestão

Nesta etapa, irá configurar o pipeline de ingestão. Cada tabela ingerida recebe uma tabela de streaming correspondente com o mesmo nome no destino. Pode usar um notebook ou a linha de comando do Databricks. Ambas as abordagens fazem chamadas API para um serviço Databricks que cria o pipeline.

Usa um caderno

O modelo no final desta página define funções auxiliares para criar e gerir o pipeline. A primeira célula define essas funções, e a segunda é onde defines o teu próprio pipeline.

1. Copia o modelo do Bloco de Notas.
2. Executa a primeira célula do caderno sem a modificar.
3. Modifique a segunda célula do notebook com os detalhes do seu pipeline (por exemplo, a tabela que deseja ingerir, onde deseja armazenar os dados, etc.).
4. Execute a segunda célula do caderno modelo; isto executa create_pipeline.
5. Podes correr list_pipeline para mostrar o id do pipeline e os seus detalhes.
6. Podes correr edit_pipeline para editar a definição do pipeline.
7. Pode executar delete_pipeline para eliminar o pipeline.

Use o CLI do Databricks

Para criar o pipeline, siga os seguintes passos:

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

Para editar a canalização:

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

Para obter a definição da linha de montagem:

databricks pipelines get "<your_pipeline_id>"

Para eliminar o pipeline:

databricks pipelines delete "<your_pipeline_id>"

Para mais informações, pode sempre executar:

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

Configurar funcionalidades adicionais (opcionais)

O conector oferece funcionalidades adicionais, como o SCD tipo 2 para rastreamento de histórico, seleção e desseleção ao nível das colunas. Veja Padrões comuns para pipelines de ingestão geridos.

Modelo de caderno

Copia ambas as células para um caderno no teu 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 defines o pipeline que queres criar.

Célula 1: Configuração da API

Copia esta célula as-is e executa-a sem alterações. Define create_pipeline, list_pipeline, edit_pipeline, delete_pipeline, e os outros ajudantes que o telemóvel 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 da quantidade de dados do seu Synapse Link que pretende ingerir:

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

Substitui os valores provisórios pelos teus, mas deixa "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 defeito, a API utiliza SCD tipo 1. Isto significa que sobrescreve os dados no destino se forem editados na fonte. Se preferir preservar dados históricos e usar SCD tipo 2, então especifique isso 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 defeito, a API ingere todas as colunas da tabela selecionada. No entanto, pode escolher 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)