Ingerir dados do Anaplan

info

Beta

Esse recurso está em Beta. Os administradores do workspace podem controlar o acesso a esse recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Esta página mostra como criar um pipeline de ingestão gerenciado do Anaplan usando o Lakeflow Connect.

Requisitos

• To create an ingestion pipeline, first meet the following requirements:

  • Seu workspace deve estar habilitado para o Unity Catalog.

  • O compute serverless deve estar habilitado para o seu workspace. Consulte Requisitos de compute serverless.

  • Para criar uma nova conexão, você deve ter os privilégios CREATE CONNECTION no metastore. Consulte Gerenciar privilégios no Unity Catalog.

    Se o conector for compatível com a criação de pipeline baseada em interface do usuário (UI), um administrador poderá criar a conexão e o pipeline ao mesmo tempo, concluindo os passos desta página. No entanto, se os usuários que criam pipelines usarem a criação de pipeline baseada em API ou forem usuários não administradores, um administrador deverá primeiro criar a conexão no Catalog Explorer. Consulte Conectar-se a fontes de ingestão gerenciadas.

  • To use an existing connection, you must have USE CONNECTION privileges or ALL PRIVILEGES on the connection object.

  • Você deve ter privilégios USE CATALOG no catálogo de destino.

  • Você deve ter os privilégios USE SCHEMA e CREATE TABLE em um esquema existente ou privilégios CREATE SCHEMA no catálogo de destino.

• Para fazer a ingestão do Anaplan, primeiro configure a autenticação a partir do Databricks e crie uma conexão. Consulte Configurar autenticação para o Anaplan e Criar uma conexão com o Anaplan.

Opções do conector

Defina opções com escopo de tabela no objeto individual. Consulte Exemplos para ver o uso.

Opção	Escopo	Obrigatório	Aplica-se a	Descrição
start_datetime	Tabela	Não	audit_events	O evento de auditoria mais antigo a ser ingerido na primeira sincronização, como um Timestamp UTC no formato ISO 8601 (por exemplo, 2026-07-09T12:00:00Z). O valor default é de 365 dias antes da primeira sincronização.

Opção	Escopo	Obrigatório	Aplica-se a	Descrição
start_datetime	Tabela	Não	audit_events	O evento de auditoria mais antigo a ser ingerido na primeira sincronização, como um Timestamp UTC no formato ISO 8601 (por exemplo, 2026-07-09T12:00:00Z). O valor default é de 365 dias antes da primeira sincronização.

Criar um pipeline de ingestão

Para obter a lista de tabelas de origem compatíveis, consulte Supported source tables.

Databricks UI

1. Na barra lateral do workspace do Databricks, clique em Ingestão de dados .
2. On the Add data page, under Databricks connectors , click Anaplan .
3. Na página Connection do assistente de ingestão, selecione a conexão que armazena suas credenciais da Anaplan. Se você tiver o privilégio CREATE CONNECTION no metastore, clique em Create connection para criar uma conexão com as credenciais de Configure authentication to Anaplan.
4. Clique em Avançar .
5. On the Ingestion setup page, enter a name for the pipeline.
6. Selecione um catálogo e um esquema para gravar os logs de eventos. Se você tiver os privilégios USE CATALOG e CREATE SCHEMA no catálogo, clique em Create schema no menu suspenso para criar um esquema.
7. Clique em Criar pipeline e continuar .
8. Na página Source , selecione as tabelas a serem ingeridas.
9. Clique em Salvar e continuar .
10. Na página Destination , selecione um catálogo e um esquema para carregar os dados. Se você tiver os privilégios USE CATALOG e CREATE SCHEMA no catálogo, clique em Criar esquema no menu suspenso para criar um esquema.
11. Clique em Salvar e continuar .
12. (Optional) On the Schedules and notifications page, click Create schedule . Set the frequency to refresh the destination tables.
13. (Opcional) Clique em Add notification para definir notificações por email para sucesso ou falha na operação do pipeline e, em seguida, clique em Save and run pipeline .

Declarative Automation Bundles

Use os Pacotes de Automação Declarativa para gerenciar pipelines do Anaplan como código. Os pacotes podem conter definições YAML de jobs e tarefas, são gerenciados usando a CLI do Databricks e podem ser compartilhados e ter execução em diferentes workspaces de destino (como desenvolvimento, staging e produção). Para obter mais informações, consulte O que são os Pacotes de Automação Declarativa?.

1. Crie um pacote usando a CLI do Databricks:

   Bash

   databricks bundle init

2. Adicione dois novos arquivos de recurso ao pacote:

   • Um arquivo de definição de pipeline (por exemplo, resources/anaplan_pipeline.yml). Consulte pipeline.ingestion_definition e Exemplos.
   • Um arquivo de definição de job que controla a frequência de ingestão de dados (por exemplo, resources/anaplan_job.yml).

3. Implante o pipeline usando a CLI do Databricks:

   Bash

   databricks bundle deploy

Databricks notebook

1. Importe o seguinte notebook para o seu Workspace do Databricks:

Abrir notebook em uma nova abaOpen in Databricks

2. Deixe as células um e dois como estão. Não modifique.

3. Modifique a terceira célula com os detalhes de configuração do seu pipeline. Consulte pipeline.ingestion_definition e Exemplos.

4. Configure opcionalmente as configurações avançadas do pipeline. Consulte Padrões comuns para pipelines de ingestão gerenciados.

5. Clique em Executar tudo .

Exemplos

O conector Anaplan disponibiliza duas tabelas de origem no esquema de origem default. Ingira tabelas individuais ou o esquema inteiro.

Ingerir tabelas específicas

Use esta opção para ingerir um subconjunto específico de tabelas ou para personalizar a nomenclatura de destino por tabela.

Declarative Automation Bundles

O seguinte arquivo de definição de pipeline ingere tabelas individuais do Anaplan:

YAML

resources:
  pipelines:
    anaplan_pipeline:
      name: anaplan_pipeline
      catalog: 'main'
      target: 'anaplan_data'
      ingestion_definition:
        connection_name: anaplan_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'audit_events'
              destination_catalog: 'main'
              destination_schema: 'anaplan_data'
              destination_table: 'audit_events'
              connector_options:
                api_source_connector_options:
                  options:
                    start_datetime: '<start-datetime>'
          - table:
              source_schema: 'default'
              source_table: 'users'
              destination_catalog: 'main'
              destination_schema: 'anaplan_data'
              destination_table: 'users'

Databricks notebook

A seguinte especificação de pipeline ingere tabelas individuais do Anaplan:

Python

pipeline_name = "anaplan_pipeline"
connection_name = "<anaplan-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "catalog": "main",
  "schema": "anaplan_data",
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "audit_events",
          "destination_catalog": "main",
          "destination_schema": "anaplan_data",
          "destination_table": "audit_events",
          "connector_options": {
            "api_source_connector_options": {
              "options": {
                "start_datetime": "<start-datetime>"
              }
            }
          }
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "users",
          "destination_catalog": "main",
          "destination_schema": "anaplan_data",
          "destination_table": "users"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Ingerir o esquema inteiro

Use esta opção para ingerir todas as tabelas de origem do Anaplan em um único esquema de destino com uma declaração.

Declarative Automation Bundles

O seguinte arquivo de definição de pipeline ingere todas as tabelas compatíveis da Anaplan em um esquema de destino:

YAML

resources:
  pipelines:
    anaplan_pipeline:
      name: anaplan_pipeline
      catalog: 'main'
      target: 'anaplan_data'
      ingestion_definition:
        connection_name: anaplan_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'anaplan_data'

Databricks notebook

A especificação do pipeline a seguir ingere todas as tabelas do Anaplan compatíveis em um esquema de destino:

Python

pipeline_name = "anaplan_pipeline"
connection_name = "<anaplan-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "catalog": "main",
  "schema": "anaplan_data",
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "schema": {
          "source_schema": "default",
          "destination_catalog": "main",
          "destination_schema": "anaplan_data"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Arquivo de definição de job do Declarative Automation Bundles

O a seguir é um exemplo de arquivo de definição de job para uso com Pacotes de Automação Declarativa. O job é executado diariamente.

Declarative Automation Bundles

YAML

resources:
  jobs:
    anaplan_job:
      name: anaplan_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: anaplan_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.anaplan_pipeline.id}

Padrões comuns

Para configurações avançadas de pipeline, consulte Padrões comuns para pipelines de ingestão gerenciados.

Os passos seguintes

Começar, programar e definir alertas no seu pipeline. Consulte Tarefas comuns de manutenção de pipeline.

Outros recursos

• FAQ do conector Anaplan
• Limitações do conector do Anaplan
• Referência do conector Anaplan
• Solucionar problemas do conector Anaplan