Ingerir dados do Celigo

info

Beta

Este recurso está em Beta. Workspace admins can control access to this recurso from the Previews page. See Gerenciar Databricks previews.

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

Requisitos

• Para criar um pipeline de ingestão, primeiro atenda aos seguintes requisitos:

  • Seu workspace deve estar habilitado para o Unity Catalog.

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

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

    If the connector supports UI-based pipeline authoring, an admin can create the connection and the pipeline at the same time by completing the os passos on this page. However, if the users who create pipelines use API-based pipeline authoring or are non-admin users, an admin must first create the connection in Catalog Explorer. See Connect to gerenciar ingestion sources.

  • 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 Celigo, primeiro configure a autenticação do Databricks e crie uma conexão. Consulte Configurar a autenticação para o Celigo e Criar uma conexão do Celigo.

Opções do conector

Defina opções com escopo de pipeline em source_configurations e 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_logs	Data e hora UTC ISO-8601 para o início do preenchimento retroativo (backfill) da primeira sincronização. Default 365 dias antes do primeiro início da sincronização.

Opção	Escopo	Obrigatório	Aplica-se a	Descrição
start_datetime	Tabela	Não	audit_logs	Data e hora UTC ISO-8601 para o início do preenchimento retroativo (backfill) da primeira sincronização. Default 365 dias antes do primeiro início da sincronização.

Criar um pipeline de ingestão

Para ver a lista de tabelas de origem compatíveis, consulte Tabelas de origem compatíveis.

Databricks UI

1. Na barra lateral do workspace do Databricks, clique em Ingestão de dados .
2. Na página Add data , em Databricks connectors , clique em Celigo .
3. Na página Connection do assistente de ingestão, selecione a conexão que armazena suas credenciais do Celigo. 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 Celigo.
4. Clique em Avançar .
5. Na página Ingestion setup , insira um nome para o 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. Click Create pipeline and continue .
8. Na página Source , selecione as tabelas para ingestão.
9. Click Save and continue .
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 Create schema no menu suspenso para criar um esquema.
11. Click Save and continue .
12. (Opcional) Na página Schedules and notifications , clique em programar . Defina a frequência para refresh das tabelas de destino.
13. (Opcional) Clique em Add notification para definir notificações por email para o 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 Celigo 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 bundle 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/celigo_pipeline.yml). Veja pipeline.ingestion_definition e Exemplos.
   • Um arquivo de definição de job que controla a frequência da ingestão de dados (por exemplo, resources/celigo_job.yml).

3. Implantar 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. Optionally configure advanced pipeline settings. See Common patterns for gerenciados ingestion pipelines.

5. Clique em Executar tudo .

Exemplos

O conector Celigo disponibiliza a tabela de origem audit_logs no esquema de origem default. Ingira a tabela diretamente ou ingira 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. Defina o início opcional do preenchimento retroativo start_datetime no objeto audit_logs.

Declarative Automation Bundles

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

YAML

resources:
  pipelines:
    celigo_pipeline:
      name: celigo_pipeline
      catalog: 'main'
      target: 'celigo_data'
      ingestion_definition:
        connection_name: celigo_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'audit_logs'
              destination_catalog: 'main'
              destination_schema: 'celigo_data'
              destination_table: 'audit_logs'
              connector_options:
                api_source_connector_options:
                  options:
                    start_datetime: '<start-datetime>'

Databricks notebook

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

Python

pipeline_name = "celigo_pipeline"
connection_name = "<celigo-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "audit_logs",
          "destination_catalog": "main",
          "destination_schema": "celigo_data",
          "destination_table": "audit_logs",
          "connector_options": {
            "api_source_connector_options": {
              "options": {
                "start_datetime": "<start-datetime>"
              }
            }
          }
        }
      }
    ]
  }
}
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 Celigo em um único esquema de destino com uma declaração.

Declarative Automation Bundles

O arquivo de definição de pipeline a seguir ingere todas as tabelas do Celigo compatíveis em um esquema de destino:

YAML

resources:
  pipelines:
    celigo_pipeline:
      name: celigo_pipeline
      catalog: 'main'
      target: 'celigo_data'
      ingestion_definition:
        connection_name: celigo_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'celigo_data'

Databricks notebook

A seguinte especificação de pipeline ingere todas as tabelas Celigo compatíveis em um esquema de destino:

Python

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

Arquivo de definição de job do Pacote de Automação Declarativa

O arquivo a seguir é um exemplo de definição de job para uso com pacotes de automação declarativa (Declarative Automation Bundles). O job é executado diariamente.

Declarative Automation Bundles

YAML

resources:
  jobs:
    celigo_job:
      name: celigo_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: celigo_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.celigo_pipeline.id}

Padrões comuns

For advanced pipeline configurations, see Common patterns for managed ingestion pipelines.

Os passos seguintes

Comece, programe e defina alertas no seu pipeline. Consulte Tarefas comuns de manutenção de pipeline.

Recursos adicionais

• Perguntas frequentes sobre o conector Celigo
• Limitações do conector Celigo
• Referência do conector Celigo
• Solucionar problemas do conector Celigo