Erfassen von Daten aus Salesforce

In diesem Artikel wird beschrieben, wie Sie Daten aus Salesforce erfassen und mithilfe von LakeFlow Connect in Azure Databricks laden. Die resultierende Erfassungspipeline wird über Unity Catalog gesteuert und von serverlosem Computing sowie Delta Live Tables (DLT) unterstützt.

Der Salesforce-Ingestion-Konnektor unterstützt die folgende Quelle:

• Salesforce Sales Cloud

Voraussetzungen

Um eine Erfassungspipeline erstellen zu können, müssen folgende Anforderungen erfüllt sein:

• Ihr Arbeitsbereich ist für Unity Catalog aktiviert.

• Serverloses Computing ist für Notebooks, Workflows und Delta Live Tables aktiviert. Weitere Informationen finden Sie unter Aktivieren des serverlosen Computings.

• Zum Erstellen eine Verbindung: Verwenden Sie CREATE CONNECTION im Metastore.

  Zum Verwenden einer vorhandenen Verbindung: Verwenden Sie USE CONNECTION oder ALL PRIVILEGES mit dem Verbindungsobjekt.

• USE CATALOG für den Zielkatalog.

• USE SCHEMA und CREATE TABLE für ein vorhandenes Schema oder CREATE SCHEMA für den Zielkatalog.

• (Empfohlen) Erstellen Sie einen Salesforce-Benutzer, den Databricks zum Abrufen von Daten verwenden kann. Stellen Sie sicher, dass der Benutzer über API-Zugriff und Zugriff auf alle Objekte verfügt, die Sie aufnehmen möchten.

Erstellen einer Salesforce-Verbindung

Erforderliche Berechtigungen: CREATE CONNECTION für den Metastore. Bitten Sie einen Metastore-Administrator, diese Berechtigung zu erteilen.

Wenn Sie eine Aufnahmepipeline mit einer vorhandenen Verbindung erstellen möchten, fahren Sie mit dem folgenden Abschnitt fort. Sie benötigen USE CONNECTION oder ALL PRIVILEGES verwenden die Verbindung.

Gehen Sie wie folgt vor, um eine Salesforce-Verbindung zu erstellen:

1. Wählen Sie im Azure Databricks-Arbeitsbereich Katalog > Externe Speicherorte > Verbindungen > Verbindung erstellen aus.

2. Geben Sie unter Verbindungsname einen eindeutigen Namen für die Salesforce-Verbindung ein.

3. Klicken Sie für Verbindungstyp auf Salesforce.

4. Wenn Sie ein Salesforce-Sandboxkonto für die Erfassung verwenden, legen Sie Ist Sandbox auf true fest.

5. Klicken Sie auf Mit Salesforce anmelden.

   

6. Wenn Sie aus einer Salesforce-Sandbox aufgenommen werden, klicken Sie auf "Benutzerdefinierte Domäne verwenden". Geben Sie die Sandkasten-URL an, und fahren Sie dann mit der Anmeldung fort. Databricks empfiehlt die Anmeldung als Salesforce-Benutzer, der sich für die Erfassung von Databricks engagiert.

   

   

7. Kehren Sie zur Seite Verbindung erstellen zurück, und klicken Sie auf Erstellen.

Erstellen einer Aufnahmepipeline

Erforderliche Berechtigungen: USE CONNECTION oder ALL PRIVILEGES für eine Verbindung.

In diesem Schritt wird beschrieben, wie Sie die Erfassungspipeline erstellen. Jeder erfassten Tabelle wird eine entsprechende Streamingtabelle mit demselben Namen (jedoch nur in Kleinbuchstaben) im Ziel zugeordnet, sofern Sie sie nicht explizit umbenannt haben.

Databricks UI

1. Klicken Sie in der Randleiste des Azure Databricks-Arbeitsbereichs auf "Datenaufnahme".

2. Klicken Sie auf der Seite "Daten hinzufügen" unter "Databricks-Connectors" auf Salesforce.

   Der Salesforce-Aufnahme-Assistent wird geöffnet.

3. Geben Sie auf der Pipelineseite des Assistenten einen eindeutigen Namen für die Aufnahmepipeline ein.

4. Wählen Sie im Dropdownmenü "Zielkatalog " einen Katalog aus. Erfasste Daten und Ereignisprotokolle werden in diesen Katalog geschrieben.

5. Wählen Sie die Unity-Katalogverbindung aus, die die anmeldeinformationen speichert, die für den Zugriff auf Salesforce-Daten erforderlich sind.

   Wenn keine Salesforce-Verbindungen vorhanden sind, klicken Sie auf "Verbindung erstellen". Sie müssen über die CREATE CONNECTION Berechtigung für den Metastore verfügen.

6. Klicken Sie auf "Pipeline erstellen", und fahren Sie fort.

7. Wählen Sie auf der Seite "Quelle " die Salesforce-Tabellen aus, die in Databricks aufgenommen werden sollen, und klicken Sie dann auf "Weiter".

   Wenn Sie ein Schema auswählen, schreibt der Salesforce-Aufnahmeconnector alle vorhandenen und zukünftigen Tabellen im Quellschema in verwaltete Tabellen im Unity-Katalog.

8. Wählen Sie auf der Seite "Ziel " den Unity-Katalog und das Schema aus, in das geschrieben werden soll.

   Wenn Sie kein vorhandenes Schema verwenden möchten, klicken Sie auf "Schema erstellen". Sie müssen über die USE CATALOG Berechtigungen für CREATE SCHEMA den übergeordneten Katalog verfügen.

9. Klicken Sie auf "Pipeline speichern", und fahren Sie fort.

10. Klicken Sie auf der Seite "Einstellungen" auf " Zeitplan erstellen". Legen Sie die Häufigkeit fest, mit der die Zieltabellen aktualisiert werden.

11. Legen Sie optional E-Mail-Benachrichtigungen für Erfolg oder Fehler des Pipelinevorgangs fest.

12. Klicken Sie auf "Speichern", und führen Sie die Pipeline aus.

Databricks-Ressourcenbundles

Auf dieser Registerkarte wird beschrieben, wie Sie eine Aufnahmepipeline mithilfe von Databricks Asset Bundles (DABs) bereitstellen. Bundles können YAML-Definitionen von Aufträgen und Aufgaben enthalten, werden über die Databricks-CLI verwaltet und können in verschiedenen Ziel-Arbeitsbereiche (z. B. Entwicklung, Staging und Produktion) gemeinsam genutzt und ausgeführt werden. Weitere Informationen finden Sie unter Databricks Asset Bundles.

1. Erstellen Eines neuen Bündels mithilfe der Databricks CLI:

   databricks bundle init
   

2. Fügen Sie dem Bundle zwei neue Ressourcendateien hinzu:

   • Eine Pipelinedefinitionsdatei (resources/sfdc_pipeline.yml).
   • Eine Workflowdatei, die die Häufigkeit der Datenaufnahme (resources/sfdc_job.yml) steuert.

   Im Folgenden finden Sie eine resources/sfdc_pipeline.yml-Beispieldatei:

   variables:
     dest_catalog:
       default: main
     dest_schema:
       default: ingest_destination_schema
   
   # The main pipeline for sfdc_dab
   resources:
     pipelines:
       pipeline_sfdc:
         name: salesforce_pipeline
         ingestion_definition:
           connection_name: <salesforce-connection>
           objects:
             # An array of objects to ingest from Salesforce. This example
             # ingests the AccountShare, AccountPartner, and ApexPage objects.
             - table:
                 source_schema: objects
                 source_table: AccountShare
                 destination_catalog: ${var.dest_catalog}
                 destination_schema: ${var.dest_schema}
             - table:
                 source_schema: objects
                 source_table: AccountPartner
                 destination_catalog: ${var.dest_catalog}
                 destination_schema: ${var.dest_schema}
             - table:
                 source_schema: objects
                 source_table: ApexPage
                 destination_catalog: ${var.dest_catalog}
                 destination_schema: ${var.dest_schema}
         channel: "preview"
   

   Im Folgenden finden Sie eine resources/sfdc_job.yml-Beispieldatei:

   resources:
     jobs:
       sfdc_dab_job:
         name: sfdc_dab_job
   
         trigger:
           # Run this job every day, exactly one day from the last run
           # See https://docs.databricks.com/api/workspace/jobs/create#trigger
           periodic:
             interval: 1
             unit: DAYS
   
         email_notifications:
           on_failure:
             - <email-address>
   
         tasks:
           - task_key: refresh_pipeline
             pipeline_task:
               pipeline_id: ${resources.pipelines.pipeline_sfdc.id}
   

3. Stellen Sie die Pipeline mithilfe der Databricks CLI bereit:

   databricks bundle deploy
   

Databricks-Befehlszeilenschnittstelle

Führen Sie zum Erstellen der Pipeline

databricks pipelines create --json "<pipeline-definition | json-file-path>"

So aktualisieren Sie das Pipelineprofil

databricks pipelines update --json "<<pipeline-definition | json-file-path>"

So rufen Sie die Pipelinedefinition ab

databricks pipelines get "<pipeline-id>"

So löschen Sie die Pipeline

databricks pipelines delete "<pipeline-id>"

Um weitere Informationen anzuzeigen, können Sie den folgenden Befehl ausführen:

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

Starten, Planen und Festlegen von Benachrichtigungen für Ihre Pipeline

1. Nachdem die Pipeline erstellt wurde, kehren Sie zum Databricks-Arbeitsbereich zurück, und klicken Sie dann auf Delta Live Tables.

   Die neue Pipeline wird in der Pipelineliste angezeigt.

2. Um die Pipelinedetails anzuzeigen, klicken Sie auf den Pipelinenamen.

3. Klicken Sie auf der Seite mit den Pipelinedetails auf Start, um die Pipeline auszuführen. Sie können die Pipeline planen, indem Sie auf Planen klicken.

4. Wenn Sie Benachrichtigungen für die Pipeline festlegen möchten, klicken Sie auf Planen, danach auf Weitere Optionen, und fügen Sie dann eine Benachrichtigung hinzu.

5. Nach Abschluss der Erfassung können Sie Ihre Tabellen abfragen.