Ingestion des données de Gmail

info

Bêta

Cette fonctionnalité est en version bêta. Les administrateurs de Workspace peuvent contrôler l’accès à cette fonctionnalité depuis la page Aperçus . Consultez Gérer les aperçus Databricks.

Créez un pipeline d’ingestion Gmail géré pour charger les messages, libellés, brouillons, filtres et profil d’une boîte aux lettres dans des tables Unity Catalog. Vous pouvez créer le pipeline dans l’interface utilisateur d’ingestion de données, avec des Declarative Automation Bundles ou via l’API Pipelines. Chaque pipeline ingère la boîte aux lettres unique définie sur la connexion.

Exigences

• Pour créer un pipeline d’ingestion, vous devez d’abord remplir les exigences suivantes :

  • Votre workspace doit être activé pour Unity Catalog.

  • Le compute serverless doit être activé pour votre workspace. Voir les exigences du compute serverless.

  • Pour créer une nouvelle connexion, vous devez disposer des privilèges CREATE CONNECTION sur le métastore. Voir Gérer les privilèges dans Unity Catalog.

    Si le connecteur prend en charge la création de pipeline basée sur l'interface utilisateur, un administrateur peut créer la connexion et le pipeline en même temps en suivant les étapes sur cette page. Cependant, si les utilisateurs qui créent des pipelines utilisent la création de pipeline basée sur API ou ne sont pas des utilisateurs administrateurs, un administrateur doit d'abord créer la connexion dans Catalog Explorer. Voir Se connecter à des sources d'ingestion gérées.

  • Pour utiliser une connexion existante, vous devez disposer des privilèges USE CONNECTION ou ALL PRIVILEGES sur l'objet de connexion.

  • Vous devez disposer des privilèges USE CATALOG sur le catalogue cible.

  • Vous devez disposer des privilèges USE SCHEMA et CREATE TABLE sur un schéma existant ou des privilèges CREATE SCHEMA sur le catalogue cible.

• Pour ingérer des données depuis Gmail, vous devez d’abord effectuer les étapes décrites dans Créer une connexion Gmail.

Créer un pipeline d'ingestion

Chaque table ingérée est écrite dans une table de streaming. Le schéma source est default. Pour consulter la liste des tables que vous pouvez ingérer, voir Tables prises en charge.

Databricks UI

1. Dans la barre latérale du workspace Databricks, cliquez sur Data Ingestion .
2. Sur la page Add data , sous Databricks connectors , cliquez sur Gmail .
3. Sur la page Connexion de l'assistant d'ingestion, sélectionnez la connexion qui stocke vos identifiants d'accès Gmail. Si vous disposez du privilège CREATE CONNECTION sur le métastore, vous pouvez cliquer sur Créer une connexion pour créer une nouvelle connexion avec les détails d'authentification dans Créer une connexion Gmail.
4. Cliquez sur Suivant .
5. Sur la page Configuration de l'ingestion , saisissez un nom unique pour le pipeline.
6. Sélectionnez un catalogue et un schéma dans lesquels écrire les logs des événements. Si vous disposez des privilèges USE CATALOG et CREATE SCHEMA sur le catalogue, vous pouvez cliquer sur Créer un schéma dans le menu déroulant pour créer un nouveau schéma.
7. Cliquez sur Créer un pipeline et continuer .
8. Sur la page Source , sélectionnez les tables à ingérer.
9. Cliquez sur Enregistrer et continuer .
10. Sur la page Destination , sélectionnez un catalogue et un schéma dans lesquels charger les données. Si vous disposez des privilèges USE CATALOG et CREATE SCHEMA sur le catalogue, vous pouvez cliquer sur Créer un schéma dans le menu déroulant pour créer un nouveau schéma.
11. Cliquez sur Enregistrer et continuer .
12. (Facultatif) Sur la page Calendriers et notifications , cliquez sur Créer un calendrier . Définissez la fréquence de refresh des tables de destination.
13. (Optionnel) Cliquez sur Ajouter une notification pour définir les notifications par e-mail pour le succès ou l’échec des opérations de pipeline, puis cliquez sur Enregistrer et exécuter le pipeline .

Declarative Automation Bundles

Utilisez les Declarative Automation Bundles pour gérer les pipelines Gmail en tant que code. Les bundles peuvent contenir des définitions YAML de Jobs et de tâches, sont gérés à l'aide de la CLI Databricks et peuvent être partagés et exécutés dans différents Workspace cibles (tels que le développement, la pré-production et la production). Pour plus d'informations, consultez Que sont les Declarative Automation Bundles ?.

1. Créez un bundle à l'aide de la CLI Databricks :

   Bash

   databricks bundle init

2. Ajoutez deux nouveaux fichiers de ressources au bundle :

   • Un fichier de définition de pipeline (par exemple, resources/gmail_pipeline.yml). Voir pipeline.ingestion_definition et Exemples.
   • Un fichier de définition de job qui contrôle la fréquence d’ingestion des données (par exemple, resources/gmail_job.yml).

3. Déploie le pipeline en utilisant la Databricks CLI :

   Bash

   databricks bundle deploy

Databricks notebook

1. Modifiez la spécification du pipeline avec les détails de votre configuration de pipeline. Voir pipeline.ingestion_definition et Exemples.
2. Exécutez le notebook.

Sélection de la boîte aux lettres

La boîte aux lettres à lire est définie sur la connexion, et non sur le pipeline. Configurez-le avec le champ E-mail de la boîte aux lettres (impersonate_email) lorsque vous créez la connexion. Voir Créer une connexion Gmail. Le compte de service usurpe l'identité de cet utilisateur via une délégation à l'échelle du domaine, et le connecteur lit la boîte aux lettres de cet utilisateur. Si vous ne le définissez pas, la connexion s'authentifie en tant que compte de service lui-même, qui ne dispose d'aucune boîte aux lettres à ingérer. Le connecteur marque la valeur de la boîte aux lettres en tant que colonne mailbox sur chaque ligne.

Chaque connexion ingère une seule boîte aux lettres. Pour ingérer plus d’une boîte aux lettres, créez une connexion et un pipeline distincts pour chaque boîte aux lettres.

Planifiez l'exécution du pipeline au moins une fois par semaine

Databricks recommande de planifier l'exécution du pipeline au moins une fois tous les sept jours. Les tables messages et message_labels se synchronisent de manière incrémentielle à l'aide de l'API History de Gmail, et Gmail conserve l'historique pendant une période limitée (généralement environ sept jours).

attention

Si le pipeline s’exécute moins fréquemment que la fenêtre d’historique de Gmail, le curseur historyId stocké peut expirer. Quand c’est le cas, la prochaine partie effectue un full refresh de messages et message_labels.

Exemples

Utilisez ces exemples pour configurer votre pipeline.

Ingérer une table source unique

Declarative Automation Bundles

Le fichier de définition de pipeline suivant ingère une seule table source. La ressource pipeline_gmail est le pipeline principal, et objects définit un tableau de tables à ingérer. Cet exemple ingère la table messages.

YAML

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

# The main pipeline for gmail_dab
resources:
  pipelines:
    pipeline_gmail:
      name: gmail_pipeline
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: <gmail-connection>
        objects:
          # An array of objects to ingest from Gmail. This example ingests the messages table.
          - table:
              source_schema: default
              source_table: messages
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

Databricks notebook

La spécification de pipeline suivante ingère une seule table source :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<gmail-connection>",
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "messages",
          "destination_catalog": "main",
          "destination_schema": "ingest_destination_schema"
        }
      }
    ]
  }
}
"""
create_pipeline(pipeline_spec)

Ingérer plusieurs tables sources

Declarative Automation Bundles

Le fichier de définition de pipeline suivant ingère plusieurs tables sources :

YAML

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

# The main pipeline for gmail_dab
resources:
  pipelines:
    pipeline_gmail:
      name: gmail_pipeline
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: <gmail-connection>
        objects:
          # An array of objects to ingest from Gmail. This example ingests the messages and message_labels tables.
          - table:
              source_schema: default
              source_table: messages
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
          - table:
              source_schema: default
              source_table: message_labels
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

Databricks notebook

La spécification de pipeline suivante ingère plusieurs tables sources :

Python

pipeline_spec = """
{
  "name": "<pipeline-name>",
  "ingestion_definition": {
    "connection_name": "<gmail-connection>",
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "messages",
          "destination_catalog": "main",
          "destination_schema": "ingest_destination_schema"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "message_labels",
          "destination_catalog": "main",
          "destination_schema": "ingest_destination_schema"
        }
      }
    ]
  }
}
"""
create_pipeline(pipeline_spec)

Fichier de définition de job Declarative Automation Bundles

Voici un exemple de fichier de définition de job à utiliser avec les Declarative Automation Bundles. Le job s'exécute tous les jours, exactement un jour après la dernière exécution.

YAML

resources:
  jobs:
    gmail_dab_job:
      name: gmail_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_gmail.id}

Étapes suivantes

start, planifiez et configurez des alertes sur votre pipeline. Voir Tâches courantes de maintenance des pipelines.

Ressources supplémentaires

• FAQ du connecteur Gmail
• Limitations du connecteur Gmail
• Référence du connecteur Gmail
• Dépanner l’ingestion de Gmail