Ingérer des données depuis Anysphere Organization

info

Bêta

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

Cette page explique comment créer un pipeline d'ingestion géré Anysphere Organization à l'aide de Lakeflow Connect.

Configuration requise

• Pour créer un pipeline d'ingestion, vous devez d'abord répondre aux exigences suivantes :

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

  • Le compute serverless doit être activé pour votre workspace. Consultez les exigences relatives au compute serverless.

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

    Si le connecteur prend en charge la création de pipelines basée sur l’interface utilisateur, un administrateur peut créer la connexion et le pipeline en même temps en effectuant les étapes de cette page. Toutefois, si les utilisateurs qui créent des pipelines utilisent une création de pipeline basée sur l’API ou ne sont pas des administrateurs, un administrateur doit d’abord créer la connexion dans Catalog Explorer. Consultez Connect to managed ingestion sources.

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

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

  • You must have USE SCHEMA and CREATE TABLE privileges on an existing schema or CREATE SCHEMA privileges on the target catalog.

• Pour procéder à l’ingestion à partir d’Anysphere Organization, vous devez d’abord configurer l’authentification depuis Databricks. Voir Configurer l’authentification pour Anysphere Organization.

Créer un pipeline d’ingestion

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 Anysphere Organization .
3. Sur la page Connection de l’assistant d’ingestion, sélectionnez la connexion qui stocke vos identifiants d’accès à l’organisation Anysphere. Si vous disposez du privilège CREATE CONNECTION sur le métastore, vous pouvez cliquer sur Create connection pour créer une nouvelle connexion avec les détails d’authentification dans Create an Anysphere Organization connection.
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 pour l'écriture des Logs d'événements. If you have USE CATALOG and CREATE SCHEMA privileges on the catalog, you can click Create schema in the drop-down menu to create a new schema.
7. Cliquez sur Create pipeline and continue .
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. (Facultatif) Cliquez sur Ajouter une notification pour configurer les notifications par e-mail en cas de réussite ou d'échec des opérations du pipeline, puis cliquez sur Enregistrer et exécuter le pipeline .

Declarative Automation Bundles

Utilisez les Declarative Automation Bundles pour gérer les pipelines Anysphere Organization 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/anysphere_organization_pipeline.yml). Voir pipeline.ingestion_definition et Exemples.
   • Fichier de définition du job qui contrôle la fréquence de l’ingestion des données (par exemple, resources/anysphere_organization_job.yml).

3. Déployez le pipeline à l'aide de la CLI Databricks :

   Bash

   databricks bundle deploy

Databricks notebook

1. Importez le notebook suivant dans votre Workspace Databricks :

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

2. Laissez les cellules un et deux telles quelles. Ne pas modifier.

3. Modifiez la troisième cellule avec les détails de configuration de votre pipeline. Voir pipeline.ingestion_definition et Exemples.

4. Vous pouvez configurer éventuellement des paramètres de pipeline avancés. See Common patterns for managed ingestion pipelines.

5. Cliquez sur Tout exécuter .

Exemples

Utilisez ces exemples pour configurer votre pipeline.

Ingérer l'ensemble du schéma

Declarative Automation Bundles

Le fichier de définition de pipeline suivant ingère toutes les tables prises en charge de l'organisation Anysphere dans un schéma de destination :

YAML

resources:
  pipelines:
    anysphere_organization_pipeline:
      name: anysphere_organization_pipeline
      catalog: 'main'
      target: 'anysphere_organization_data'
      ingestion_definition:
        connection_name: anysphere_organization_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'

Databricks notebook

La spécification de pipeline suivante ingère toutes les tables prises en charge de l'organisation Anysphere dans un schéma de destination :

Python

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

Ingérer des tables spécifiques

Ingérez des tables individuelles de l'organisation Anysphere. Pour obtenir la liste complète des tables prises en charge, consultez la section Supported source tables.

Declarative Automation Bundles

The following pipeline definition file ingests individual Anysphere Organization tables:

YAML

resources:
  pipelines:
    anysphere_organization_pipeline:
      name: anysphere_organization_pipeline
      catalog: 'main'
      target: 'anysphere_organization_data'
      ingestion_definition:
        connection_name: anysphere_organization_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'organization_groups'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_groups'
          - table:
              source_schema: 'default'
              source_table: 'organization_group_members'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_group_members'
          - table:
              source_schema: 'default'
              source_table: 'organization_members'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_members'

Databricks notebook

La spécification de pipeline suivante ingère des tables d'organisation Anysphere individuelles :

Python

pipeline_name = "anysphere_organization_pipeline"
connection_name = "<anysphere-organization-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_groups",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_groups"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_group_members",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_group_members"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_members",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_members"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Fichier de définition de job Declarative Automation Bundles

Declarative Automation Bundles

YAML

resources:
  jobs:
    anysphere_organization_job:
      name: anysphere_organization_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: anysphere_organization_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.anysphere_organization_pipeline.id}

Modèles courants

Pour les configurations de pipeline avancées, consultez la section Common patterns for managed ingestion pipelines.

Étapes suivantes

start, planifiez et configurez des alertes sur votre pipeline. Voir les tâches de maintenance courantes du pipeline.

Ressources supplémentaires

• Limitations des connecteurs de l’organisation Anysphere
• Référence du connecteur Anysphere Organization
• FAQ sur le connecteur Anysphere Organization