Configurar fonte de dados para ingestão Microsoft Dynamics 365

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

nota

Esta página aborda o fluxo de trabalho de exportação CSV, que não usa um workspace do Azure Synapse Analytics. Para exportar como tabelas Delta no formato Parquet usando um workspace do Azure Synapse Analytics, consulte Configurar uma fonte de dados Parquet para ingestão do Microsoft Dynamics 365. A Databricks recomenda o fluxo de trabalho Parquet para instâncias grandes ou de alto volume, pois ele oferece melhor desempenho e estabilidade em escala.

Para obter informações sobre como o conector acessa seus dados de origem, consulte Como o conector acessa os dados do D365?. Para obter uma lista dos aplicativos Dataverse compatíveis, consulte Quais aplicativos do Dynamics 365 são compatíveis?.

Pré-requisitos

Antes de configurar a fonte de dados do Dynamics 365, é necessário ter:

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

Configure 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 que não são do Dataverse, você deve configurar entidades virtuais ou tabelas diretas antes de configurar o Azure Synapse Link.

Para configurar entidades virtuais:

1. No Power Apps , acesse a página Ambientes e clique em Aplicativos do Dynamics 365 .

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

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

4. Para cada entidade virtual que você deseja importar, habilite o recurso "Controlar alterações" em "Propriedades avançadas" .

5. Por default, a solução F&O Virtual Entity default algumas entidades virtuais na lista de tabelas do Dataverse. No entanto, você pode expor entidades adicionais manualmente:

   1. Acesse a página de Configurações Avançadas do seu ambiente Dataverse.
   2. Clique no ícone de filtro no canto superior direito para acessar a pesquisa avançada.
   3. Selecione "Entidades de Financiamento e Operações Disponíveis" no menu suspenso e clique em "Resultados" .
   4. Selecione a entidade virtual que deseja expor.
   5. Na página de administração da entidade , alterne a opção Visível para Verdadeiro 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_.

importante

Entidades virtuais e tabelas diretas aparecem no Azure Synapse Link somente após o Dataverse terminar de sincronizá-las. Isso geralmente leva até 15 minutos, mas pode levar até 30. Se as tabelas estiverem ausentes após 30 minutos, consulte Virtual entities not appearing in schema discovery.

Configurar o Azure Synapse Link

Neste passo, você usará o Synapse Link for Dataverse to Azure Data Lake para escolher as tabelas que deseja ingerir. Este serviço substitui o serviço anteriormente conhecido como Export data to 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, em Vincular ao Azure Synapse .

2. Clique em Novo Link . O Dataverse preenche automaticamente suas inscrições ativas do mesmo tenant. Selecione a inscrição apropriada no dropdown.

3. Não selecione a caixa de seleção Connect to your Azure Synapse Analytics Workspace . Os dados são armazenados como CSV diretamente na sua account de armazenamento ADLS Gen2, e este fluxo de trabalho não requer um workspace do Azure Synapse Analytics.

4. Na página de criação de links do Synapse , clique em Avançado . Em seguida, ative a opção Mostrar configurações avançadas .

5. Ative a opção "Atualizar estrutura de pastas incrementalmente" e defina o intervalo de atualização desejado para o Synapse Link. O mínimo é de 5 minutos. Este intervalo aplica-se a todas as tabelas incluídas neste link do Synapse. (Você definirá um programador para seu pipeline Databricks em uma etapa separada.)

6. Selecione as tabelas que você deseja sincronizar, deixando as configurações Append only e Partition como default.

   • Se estiver importando dados de um aplicativo nativo do Dataverse, selecione as tabelas relevantes do Dataverse diretamente na seção Dataverse .
   • Ao importar dados do F&O, você pode selecionar tabelas diretas da seção D365 Finance & operações ou entidades virtuais da seçã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 usuários de F&O, essa sincronização inicial pode levar horas para tabelas grandes com centenas de gigabytes.

nota

Se a sincronização inicial de uma entidade F&O demorar muito, você pode acelerá-la criando um índice na tabela no aplicativo F&O:

1. Navegue até a tabela que deseja indexar no ambiente F&O.
2. Crie uma extensão para a tabela.
3. Dentro da extensão da tabela, defina um novo índice.
4. Adicione os campos que deseja incluir no índice, o que acelera as pesquisas de banco de dados nesses campos.
5. Salve e implemente as alterações no seu ambiente de F&O.

Crie um aplicativo Entra ID para ingestão.

Nesta etapa, você coletará as informações do ID de entrada necessárias para criar uma conexão com Unity Catalog que suporte a ingestão no Databricks.

1. Colete o IDtenant do seu tenant do Entra ID (portal.azure.com >> Microsoft Entra ID >> tabVisão geral >> ID do locatário , listado no painel direito).

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

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

   1. Crie um aplicativo Microsoft Entra ID, caso ainda não tenha um.
   2. Recolha o segredo do cliente .
   3. Colete o ID do aplicativo (portal.azure.com >> IDMicrosoft Entra >> gerenciar >> Registros de aplicativos ).

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

nota

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

1. Acesse sua conta de armazenamentoAzure e selecione seu contêiner ou account de armazenamento. (A Databricks recomenda o nível do contêiner para manter o princípio do menor privilégio.)
2. Clique em Controle de Acesso (IAM) e, em seguida, em Adicionar atribuição de função .
3. Selecione a função de acesso Contribuidor de Dados do Blob de Armazenamento → Leitura/Gravação/Exclusão. Se a sua organização não permitir isso, entre em contato com a equipe da sua account Databricks .
4. Clique em Avançar e, em seguida, selecione Membros .
5. Selecione Usuário, grupo ou entidade de serviço e, em seguida , Pesquise o registro do seu aplicativo . (Caso o aplicativo não esteja presente nos resultados da pesquisa, você pode inserir explicitamente o ID do objeto na barra de pesquisa e pressionar Enter ).
6. Clique em Revisar + Atribuir .
7. Para confirmar que as permissões estão configuradas corretamente, você pode verificar o Controle de Acesso do seu contêiner.

Criar um pipeline do Dynamics 365

Você pode criar o pipeline na interface do usuário ou por meio da API. O assistente da interface do usuário lida com a conexão e o pipeline juntos, enquanto o caminho da API os cria como dois os passos separados.

Use a interface do usuário

O assistente solicita as credenciais do aplicativo Entra ID e os detalhes de armazenamento que você reuniu n'os passos anteriores e, em seguida, cria a conexão e o pipeline juntos.

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

Utilize a API

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

Etapa 1: criar uma conexão do Dynamics 365

Neste passo, será criada uma conexão do Unity Catalog para armazenar com segurança suas credenciais do Dynamics 365 e iniciar a ingestão no Databricks.

1. Na sua workspace, clique em Catálogo .
2. Clique Conecte-se e, em seguida, clique em Conexões .
3. Clique no botão Criar conexão .
4. Forneça um nome de conexão exclusivo e, em seguida, selecione Dynamics 365 como o tipo de conexão .
5. Insira o segredo do cliente e o ID do cliente do aplicativo Entra ID criado na etapa anterior. Não modifique o escopo. Clique em Avançar .
6. Insira o nome da conta do ArmazenamentoAzure , o ID do locatário e o nome do contêinerADLS e clique em Criar conexão.
7. Anote o nome da conexão.

O passo 2: Criar o pipeline de ingestão

Neste passo, você configurará o pipeline de ingestão. Cada tabela ingerida recebe uma tabela de transmissão correspondente com o mesmo nome no destino. Você pode usar um Notebook ou a CLI do Databricks. Ambas as abordagens fazem chamadas de API para um serviço do Databricks que cria o pipeline.

Usar um notebook

O padrão no 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 Notebook padrão.
2. executar a primeira célula do Notebook sem modificá-la.
3. Modifique a segunda célula do Notebook com os detalhes do seu pipeline (por exemplo, a tabela da qual você deseja ingerir os dados, onde deseja armazená-los, etc.).
4. execução da segunda célula do Caderno padrão; esta execução create_pipeline.
5. Você pode executar list_pipeline para mostrar o ID pipeline e seus detalhes.
6. Você pode executar edit_pipeline para editar a definição pipeline .
7. Você pode executar delete_pipeline para excluir o pipeline.

Use a CLI do Databricks

Para criar o pipeline:

Bash

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

Para editar o pipeline:

Bash

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

Para obter a definição do pipeline:

Bash

databricks pipelines get "<your_pipeline_id>"

Para excluir o pipeline:

Bash

databricks pipelines delete "<your_pipeline_id>"

Para mais informação pode sempre executar:

Bash

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

Configurar recurso adicional (opcional)

O conector oferece recursos adicionais, como SCD tipo 2 para história acompanhamento, seleção e deseleção em nível de coluna. Consulte Padrões comuns para gerenciar o pipeline de ingestão.

Notebook

Copie ambas as células para um notebook em seu workspace. A célula 1 define as funções auxiliares que chamam a API de pipelines, e a célula 2 é onde você define o pipeline que deseja criar.

Célula 1: configuração da API

Copie esta célula como está e execute-a sem alterações. Ele define create_pipeline, list_pipeline, edit_pipeline, delete_pipeline e os outros auxiliares que a célula 2 chama.

Python

# 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 de pipeline

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

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

Substitua os valores temporários pelos seus, mas mantenha "channel": "PREVIEW" como está.

Python

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

Python

# 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: rastrear histórico com SCD tipo 2

Por default, a API usa o tipo SCD 1. Isso significa que ele sobrescreve os dados no destino se eles forem editados na origem. Se preferir preservar os dados históricos e usar o tipo 2 SCD , especifique isso na configuração. Por exemplo:

Python

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

Python

# 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 default, a API importa todas as colunas da tabela selecionada. No entanto, você pode optar por incluir ou excluir colunas específicas. Por exemplo:

Python

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