Ingerir dados do Salesforce Marketing Cloud

info

Beta

Este recurso está em Beta. Os administradores do espaço de trabalho 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 Salesforce Marketing Cloud usando o Lakeflow Connect.

Requisitos

• Para criar um pipeline de ingestão, você deve primeiro atender aos seguintes requisitos:

  • Seu workspace deve estar habilitado para o Unity Catalog.

  • O compute serverless deve ser habilitado para seu workspace. Consulte Requisitos de computação serverless.

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

    Se o conector suportar a criação de pipelines baseada na IU, um administrador poderá criar a conexão e o pipeline simultaneamente, ao concluir os passos nesta página. No entanto, se os usuários que criam pipelines usam a autoria de pipeline baseada em API ou são usuários não administradores, um administrador deve primeiro criar a conexão no Catalog Explorer. Consulte Conectar-se a fontes de ingestão gerenciadas.

  • Para usar uma conexão existente: é preciso ter os privilégios USE CONNECTION ou ALL PRIVILEGES no objeto de conexão.

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

  • É necessário ter os privilégios USE SCHEMA e CREATE TABLE em um esquema existente ou os privilégios CREATE SCHEMA no catálogo de destino.

• Para ingerir do Salesforce Marketing Cloud, você deve concluir Configure o OAuth para ingestão do Salesforce Marketing Cloud no Databricks e crie uma conexão. Consulte Crie uma conexão do Salesforce Marketing Cloud.

Criar pipeline de ingestão

Databricks UI

1. Na barra lateral do workspace do Databricks, clique em Ingestão de dados .
2. Na página Adicionar dados , em Conectores do Databricks , clique em Salesforce Marketing Cloud .
3. Na página Conexão do assistente de ingestão, selecione a conexão que armazena suas credenciais de acesso do Salesforce Marketing Cloud. Se você tiver o privilégio CREATE CONNECTION no metastore, poderá clicar em Criar conexão para criar uma nova conexão com os detalhes de autenticação em Criar uma conexão do Salesforce Marketing Cloud.
4. Clique em Avançar .
5. Na página de Configuração de Ingestão , insira um nome exclusivo para o pipeline.
6. Selecione um catálogo e um esquema onde os logs de eventos serão gravados. Se houver os privilégios USE CATALOG e CREATE SCHEMA no catálogo, pode-se clicar em Criar esquema no menu suspenso para criar um novo esquema.
7. Clique em **Criar pipeline e continuar**.
8. Na **Página de Origem**, selecione as tabelas para ingestão.
9. Clique em Salvar e continuar .
10. Na página Destino , selecione um catálogo e um esquema para carregar dados. Se houver os privilégios USE CATALOG e CREATE SCHEMA no catálogo, pode-se clicar em Criar esquema no menu suspenso para criar um novo esquema.
11. Clique em Salvar e continuar .
12. (Opcional) Na página Cronogramas e notificações , clique em Crie um agendamento . Defina a frequência para fazer o refresh das tabelas de destino.
13. (Opcional) Clique em Adicionar notificação para configurar notificações por email para sucesso ou falha da operação do pipeline e, em seguida, clique em Salvar e executar pipeline .

Declarative Automation Bundles

Use Pacotes de Automação Declarativa para gerenciar pipelines do Salesforce Marketing Cloud 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 executados em diferentes workspaces de destino (como desenvolvimento, preparo e produção). Para obter mais informações, consulte O que são Pacotes de Automação Declarativa?.

1. Crie um pacote utilizando 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/sfmc_pipeline.yml). Veja pipeline.definição_de_ingestão e Exemplos.
   • Um arquivo de definição de Job que controla a frequência da ingestão de dados (por exemplo, resources/sfmc_job.yml).

3. Implante o pipeline usando a CLI do Databricks:

   Bash

   databricks bundle deploy

Databricks notebook

1. Importe o seguinte notebook em seu workspace do Databricks:

Abrir notebook em uma nova abaOpen in Databricks

2. Manter a célula um como está.

3. Modifique a célula dois ou três com os detalhes de configuração do seu pipeline. Veja Exemplos.

4. Clique em Executar tudo .

Exemplos

Use estes exemplos para configurar seu pipeline.

Ingerir todas as tabelas do esquema default

Declarative Automation Bundles

O seguinte arquivo de definição de pipeline ingere todas as tabelas atuais e futuras do esquema default:

YAML

variables:
  dest_catalog:
    default: main
  dest_schema:
    default: ingest_destination_schema

resources:
  pipelines:
    pipeline_sfmc:
      name: sfmc_pipeline
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: <sfmc-connection>
        objects:
          # Ingest all tables from the default schema
          - schema:
              source_schema: default
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

Databricks notebook

A seguinte especificação de pipeline ingere todas as tabelas atuais e futuras do esquema default:

Python

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

Ingerir uma única tabela de origem

Declarative Automation Bundles

O seguinte arquivo de definição do pipeline ingere uma única tabela do esquema default:

YAML

variables:
  dest_catalog:
    default: main
  dest_schema:
    default: ingest_destination_schema

resources:
  pipelines:
    pipeline_sfmc:
      name: sfmc_pipeline
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: <sfmc-connection>
        objects:
          # Ingest the campaigns table from the default schema
          - table:
              source_schema: default
              source_table: campaigns
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

Databricks notebook

A seguinte especificação de pipeline ingere uma única tabela do esquema default:

Python

pipeline_name = "sfmc_pipeline"
connection_name = "<sfmc-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "campaigns",
          "destination_catalog": "main",
          "destination_schema": "ingest_destination_schema"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Ingerir extensões de dados

Para ingerir dados de uma extensão de dados, use a External Key como valor de source_table. Consulte conceitos do conector do Salesforce Marketing Cloud para obter detalhes sobre os tipos de esquema de origem.

Declarative Automation Bundles

O seguinte arquivo de definição de pipeline ingere uma extensão de dados:

YAML

variables:
  dest_catalog:
    default: main
  dest_schema:
    default: ingest_destination_schema

resources:
  pipelines:
    pipeline_sfmc:
      name: sfmc_pipeline
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: <sfmc-connection>
        objects:
          # Ingest a data extension using its External Key (Customer Key)
          - table:
              source_schema: dataextensions
              source_table: <external-key>
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

Databricks notebook

A seguinte especificação de pipeline ingere uma extensão de dados:

Python

pipeline_name = "sfmc_pipeline"
connection_name = "<sfmc-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "dataextensions",
          "source_table": "<external-key>",
          "destination_catalog": "main",
          "destination_schema": "ingest_destination_schema"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Arquivo de definição de Job de Pacotes de Automação Declarativa

Declarative Automation Bundles

A seguir, um exemplo de arquivo de definição de job para usar com Pacotes de Automação Declarativa. O job é executado todos os dias, exatamente um dia após a última execução.

YAML

resources:
  jobs:
    sfmc_dab_job:
      name: sfmc_dab_job

      trigger:
        periodic:
          interval: 1
          unit: DAYS

      email_notifications:
        on_failure:
          - <email-address>

      tasks:
        - task_key: refresh_pipeline
          pipeline_task:
            pipeline_id: ${resources.pipelines.pipeline_sfmc.id}

Padrões comuns

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

Passos seguintes

Iniciar, programar e definir alertas no seu pipeline. Veja Tarefas comuns de manutenção de pipelines.

Recursos adicionais

• Limitações do conector do Salesforce Marketing Cloud
• Solucionar problemas do conector do Salesforce Marketing Cloud
• Referência do conector do Salesforce Marketing Cloud