Configurer l'ingestion incrémentielle depuis Amazon S3

Cet article décrit comment configurer l'ingestion de données incrémentielle d'Amazon S3 vers Databricks. Vous apprendrez à accéder en toute sécurité aux données source dans un emplacement de stockage d’objets cloud qui correspond à un volume Unity Catalog (recommandé) ou à un emplacement externe Unity Catalog. Ensuite, vous apprendrez comment ingérer les données de manière incrémentielle dans une table gérée par Unity Catalog à l'aide d'Auto Loader avec LakeFlow Pipelines.

remarque

Pour configurer l'ingestion incrémentielle dans Databricks SQL plutôt que dans un notebook, consultez Utiliser les tables de streaming autonomes.

Avant de commencer

Si vous n'êtes pas administrateur, cet article suppose qu'un administrateur vous a fourni ce qui suit :

• Accès à un workspace Databricks avec Unity Catalog activé. Pour plus d'informations, consultez Se familiariser avec Unity Catalog.

• L'autorisation READ VOLUME sur le volume externe Unity Catalog ou l'autorisation READ FILES sur l'emplacement externe Unity Catalog qui correspond à l'emplacement de stockage cloud contenant vos données source. Pour plus d'informations, consultez Accorder des autorisations sur un emplacement externe

• Le chemin d'accès à vos données source.

  Exemple de chemin de volume : /Volumes/<catalog>/<schema>/<volume>/<path>/<folder>

  Exemple de chemin d’accès à un emplacement externe : s3://<bucket>/<folder>/

• Les privilèges USE SCHEMA et CREATE TABLE sur le schéma dans lequel vous souhaitez charger des données.

• Autorisation de création de cluster ou accès à une stratégie de cluster qui définit un cluster de pipeline (champ cluster_type défini sur dlt).

  Si le chemin d'accès à vos données source est un chemin de volume, votre cluster doit exécuter Databricks Runtime 13.3 LTS ou version ultérieure.

important

Si vous avez des questions concernant ces prérequis, contactez l'administrateur de votre compte.

Étape 1 : Créer un cluster

Pour créer un cluster, procédez comme suit :

1. Connectez-vous à votre Databricks Workspace.
2. Dans la barre latérale, cliquez sur Nouveau > Cluster .
3. Dans l'interface utilisateur des clusters, spécifiez un nom unique pour votre cluster.
4. Si le chemin d'accès à vos données source est un chemin de volume, pour la version de Databricks Runtime , sélectionnez 13.2 ou une version ultérieure.
5. Cliquez sur Créer un cluster .

Étape 2 : Créer un Notebook d'exploration de données

Cette section explique comment créer un Notebook d'exploration de données afin que vous puissiez comprendre vos données avant de créer votre pipeline de données.

1. Dans la barre latérale, cliquez sur +Nouveau > Notebook .

   Le Notebook est automatiquement attaché au dernier cluster que vous avez utilisé (dans ce cas, le cluster que vous avez créé à l' Étape 1 : Créer un cluster ).

2. Saisissez un nom pour le notebook.

3. Cliquez sur le bouton de langue, puis sélectionnez Python ou SQL dans le menu déroulant. Python est sélectionné par default.

4. Pour confirmer l'accès aux données de votre source de données dans S3, collez le code suivant dans une cellule de Notebook, cliquez sur , puis cliquez sur Exécuter la cellule .

SQL

SQL

LIST '<path-to-source-data>'

Python

Python

%fs ls '<path-to-source-data>'

Remplacez <path-to-source-data> par le chemin d’accès au répertoire qui contient vos données.

Ceci affiche le contenu du répertoire qui contient le dataset.

5. Pour afficher un échantillon des enregistrements afin de mieux comprendre le contenu et le format de chaque enregistrement, collez ce qui suit dans une cellule de Notebook, cliquez sur, puis cliquez sur **Exécuter la cellule**.

SQL

SQL

SELECT * from read_files('<path-to-source-data>', format => '<file-format>') LIMIT 10

Python

Python

spark.read.format('<file-format>').load('<path-to-source-data>').limit(10).display()

Remplacez les valeurs suivantes :

• <file-format>: un format de fichier pris en charge. Voir les options de DataFrameReader.
• <path to source data>: Le chemin d'accès à un fichier dans le répertoire qui contient vos données.

Ceci affiche les dix premiers enregistrements du fichier spécifié.

Étape 3 : Ingérer les données brutes

Pour ingérer des données brutes, procédez comme suit :

1. Dans la barre latérale, cliquez sur Nouveau > Notebook .

   Le Notebook est automatiquement associé au dernier cluster que vous avez utilisé (dans ce cas, le cluster que vous avez créé précédemment dans cet article).

2. Saisissez un nom pour le notebook.

3. Cliquez sur le bouton de langue, puis sélectionnez Python ou SQL dans le menu déroulant. Python est sélectionné par default.

4. Collez le code suivant dans une cellule de Notebook :

SQL

SQL

CREATE OR REFRESH STREAMING TABLE
  <table-name>
AS SELECT
  *
FROM
  STREAM read_files(
    '<path-to-source-data>',
    format => '<file-format>'
  )

Python

Python

@dp.table(table_properties={'quality': 'bronze'})
def <table-name>():
  return (
     spark.readStream.format('cloudFiles')
     .option('cloudFiles.format', '<file-format>')
     .load(f'{<path-to-source-data>}')
 )

Remplacez les valeurs suivantes :

• <table-name>: Nom de la table qui contiendra les enregistrements ingérés.
• <path-to-source-data>: Le chemin d’accès à vos données source.
• <file-format>: un format de fichier pris en charge. Voir les options de DataFrameReader.

remarque

Les LakeFlow Pipelines ne sont pas conçus pour s'exécuter de manière interactive dans les cellules de notebook. L'exécution d'une cellule qui contient la syntaxe LakeFlow Pipelines dans un Notebook renvoie un message indiquant si la query est syntaxiquement valide, mais n'exécute pas la logique de la query. L'étape suivante décrit comment créer un pipeline à partir du notebook d'ingestion que vous venez de créer.

Étape 4 : Créer et publier un pipeline

Pour créer un pipeline et le publier dans Unity Catalog, procédez comme suit :

1. Dans votre Workspace, cliquez sur **Tâches et pipelines** dans la barre latérale.
2. Sous Nouveau , cliquez sur pipeline ETL .
3. Saisissez un nom pour votre pipeline.
4. Pour le mode pipeline , sélectionnez Trigger .
5. Pour le code source , sélectionnez le Notebook qui contient le code source de votre pipeline.
6. Pour **Destination**, sélectionnez **Unity Catalog**.
7. Pour vous assurer que votre table est gérée par Unity Catalog et que tout utilisateur ayant accès au schéma parent peut la query, sélectionnez un Catalogue et un Schéma cible dans les listes déroulantes.
8. Si vous n'avez pas l'autorisation de créer un cluster, sélectionnez une politique de cluster qui prend en charge les Lakeflow Pipelines dans la liste déroulante.
9. Pour **Avancé**, définissez le **Canal de distribution** sur **Aperçu**.
10. Acceptez toutes les autres valeurs default et cliquez sur **Créer**.

Étape 5 : Planifier le pipeline

Pour planifier le pipeline, procédez comme suit :

1. Dans votre Workspace, cliquez sur **Tâches et pipelines** dans la barre latérale.
2. Cliquez sur le nom du pipeline que vous souhaitez planifier.
3. Cliquez sur Planifier > Ajouter une planification .
4. Pour le Nom du Job , entrez un nom pour le Job.
5. Définissez le Calendrier sur Planifié .
6. Spécifiez la période, l'heure de début et le fuseau horaire.
7. Configurez une ou plusieurs adresses e-mail pour recevoir des alertes en cas de start, de succès ou d'échec du pipeline.
8. Cliquez sur Créer .

Étapes suivantes

• Accordez aux utilisateurs l'accès à la nouvelle table. Pour plus d'informations, consultez la référence des privilèges Unity Catalog.
• Les utilisateurs ayant accès à la nouvelle table peuvent désormais interroger la table dans un notebook ou utiliser l'éditeur Databricks SQL.