Utiliser des données non structurées dans des volumes

Cette page vous montre comment stocker, interroger et traiter des fichiers de données non structurés à l’aide de volumes de catalogue Unity. Vous allez apprendre à charger des fichiers, interroger des métadonnées, traiter des fichiers avec des fonctions IA, appliquer le contrôle d’accès et partager des volumes avec d’autres organisations. Si possible, des instructions pour l’utilisation de ce didacticiel à l’aide de l’interface utilisateur de l’Explorateur de catalogues ont été incluses. Si aucune option de l’Explorateur de catalogue n’est affichée, utilisez les commandes Python ou SQL fournies.

Pour obtenir une vue d’ensemble complète des fonctionnalités de volume et des cas d’usage, consultez Qu’est-ce que les volumes catalogue Unity ?.

Note

Ce tutoriel utilise des fonctions d’IA pour traiter les fichiers par chemin. Disponible en version Bêta, ce FILE type permet de stocker des références de fichiers et des métadonnées sous forme de valeurs de colonnes dans un tableau. Voir le type de fichier et les données non structurées.

Spécifications

  • Un espace de travail Azure Databricks avec Unity Catalog activé.
  • CREATE CATALOG privilège sur le metastore. Consultez Créer des catalogues. Si vous ne pouvez pas créer de catalogue, demandez à votre administrateur d’accéder ou d’utiliser un catalogue existant où vous disposez du CREATE SCHEMA privilège.
  • Databricks Runtime 14.3 LTS et versions ultérieures.
  • Pour les fonctions IA : espace de travail dans une région prise en charge.
  • Pour OpenSharing : privilèges CREATE SHARE et CREATE RECIPIENT sur le metastore. Consultez Partager des données et des ressources IA en toute sécurité.

Étape 1 : Créer un volume

Créez un catalogue, un schéma et un volume pour stocker vos fichiers. Pour obtenir des instructions détaillées sur la gestion des volumes, consultez Créer et gérer des volumes de catalogue Unity.

Étape 1.1 : Créer un catalogue et un schéma

SQL

-- Create a catalog
CREATE CATALOG IF NOT EXISTS unstructured_data_lab;
USE CATALOG unstructured_data_lab;

-- Create a schema
CREATE SCHEMA IF NOT EXISTS raw;
USE SCHEMA raw;

Python

spark.sql("CREATE CATALOG IF NOT EXISTS unstructured_data_lab")
spark.sql("USE CATALOG unstructured_data_lab")
spark.sql("CREATE SCHEMA IF NOT EXISTS raw")
spark.sql("USE SCHEMA raw")

Explorateur de catalogues

  1. Cliquez sur l’icône Données.Catalogue dans la barre latérale.
  2. Cliquez sur Créer>un catalogue.
  3. Entrez unstructured_data_lab comme nom du catalogue.
  4. Cliquez sur Créer.
  5. Cliquez sur Afficher le catalogue.

Dans la page du catalogue :

  1. Cliquez sur Créer un schéma.
  2. Entrez brut en tant que nom du schéma.
  3. Cliquez sur Créer.

Étape 1.2 : Créer un volume managé

SQL

CREATE VOLUME IF NOT EXISTS files_volume
COMMENT 'Volume for storing unstructured data files';

Python

spark.sql("""
    CREATE VOLUME IF NOT EXISTS files_volume
    COMMENT 'Volume for storing unstructured data files'
""")

Explorateur de catalogues

Sur la page de schéma :

  1. Cliquez sur Créer un>volume.
  2. Entrez files_volume comme nom de volume.
  3. Vérifiez que le volume managé est sélectionné.
  4. Cliquez sur Créer.

Étape 2 : Charger des fichiers

Chargez des fichiers dans votre volume. Pour obtenir des exemples de gestion de fichiers complets, consultez Utiliser des fichiers dans des volumes de catalogue Unity.

Étape 2.1 : Charger des fichiers

Vous pouvez utiliser des exemples à partir de databricks-datasets ce didacticiel ou charger vos propres fichiers à l’aide de l’interface utilisateur de l’Explorateur de catalogues.

Note

Vous pouvez utiliser les commandes Python pour copier des fichiers depuis databricks-datasets votre volume même si vous n’êtes pas familiarisé avec Python. Consultez Gérer les notebooks Databricks pour obtenir des instructions sur l’exécution de commandes dans les notebooks.

Python

# Upload a single image file
dbutils.fs.cp(
    "dbfs:/databricks-datasets/flower_photos/roses/10090824183_d02c613f10_m.jpg",
    "/Volumes/unstructured_data_lab/raw/files_volume/rose.jpg"
)

# Upload a single PDF file
dbutils.fs.cp(
    "dbfs:/databricks-datasets/COVID/CORD-19/2020-03-13/COVID.DATA.LIC.AGMT.pdf",
    "/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf"
)

# Upload a directory
local_dir = "dbfs:/databricks-datasets/samples/data/mllib"
volume_path = "/Volumes/unstructured_data_lab/raw/files_volume/sample_files"

for file_info in dbutils.fs.ls(local_dir):
    source = file_info.path
    dest = f"{volume_path}/{file_info.name}"
    dbutils.fs.cp(source, dest, recurse=True)
    print(f"Uploaded: {file_info.name}")

Explorateur de catalogues

Le code Python de l’onglet Python charge deux fichiers (un JPG et un PDF) et un répertoire qui inclut des fichiers .txt et .csv. Pour charger des fichiers à l’aide de l’Explorateur de catalogues :

  1. Depuis la page du volume, cliquez sur Charger sur ce volume.
  2. Dans la boîte de dialogue Charger des fichiers , sous Fichiers, cliquez sur Parcourir ou glisser-déplacer des fichiers dans la zone de dépôt.
  3. Sous Volume de destination, vérifiez que le volume que vous avez créé à l’étape précédente est sélectionné.

Étape 2.2 : Vérifier le chargement

SQL

LIST '/Volumes/unstructured_data_lab/raw/files_volume/';

Python

files = dbutils.fs.ls("/Volumes/unstructured_data_lab/raw/files_volume/")
for f in files:
    print(f"{f.name}\t{f.size} bytes")

Explorateur de catalogues

Lorsque des fichiers sont chargés, ils apparaissent sur la page du volume. Cliquez sur un nom de fichier pour afficher un aperçu, ou cliquez sur un répertoire pour afficher des fichiers individuels.

Alternative : Utiliser la commande magique %fs

Utilisez la %fs commande magique :

%fs ls /Volumes/unstructured_data_lab/raw/files_volume/

Étape 3 : Interroger les métadonnées du fichier

Interrogez les informations de fichier pour comprendre ce qui se trouve dans votre volume. Pour plus d’informations sur les modèles d’interrogation, consultez Lister et interroger des fichiers dans des volumes avec SQL.

Étape 3.1 : Afficher les métadonnées du fichier

SQL

SELECT
  path,
  _metadata.file_name,
  _metadata.file_size,
  _metadata.file_modification_time
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile'
);

Python

df = (
    spark.read
    .format("binaryFile")
    .option("recursiveFileLookup", "true")
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")
)

df.select("path", "modificationTime", "length").show(truncate=False)

Explorateur de catalogues

La page de volume de l’Explorateur de catalogue affiche le nom de chaque fichier (y compris l’extension ), la taille et la date de dernière modification .

Étape 4 : Interroger et traiter des fichiers

Utilisez les fonctions IA Azure Databricks pour extraire du contenu à partir de documents et analyser des images. Pour obtenir une vue d’ensemble complète des fonctionnalités de fonction IA, consultez Enrichir des données à l’aide d’AI Functions.

Note

Les fonctions IA nécessitent un espace de travail dans une région prise en charge. Consultez Enrichir des données à l’aide d’AI Functions.

Si vous n’avez pas accès aux fonctions IA, utilisez plutôt des bibliothèques Python standard. Développez les sections alternatives ci-dessous pour obtenir des exemples.

Étape 4.1 : Analyser des documents

SQL

SELECT
  path AS file_path,
  ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.pdf'
);

Python

result_df = spark.sql("""
    SELECT
      path AS file_path,
      ai_parse_document(content, map('version', '2.0')) AS parsed_content
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.pdf'
    )
""")
display(result_df)
Alternative : Analyser les fichiers PDF sans fonctions IA

Si les fonctions IA ne sont pas disponibles dans votre région, utilisez des bibliothèques Python :

%pip install PyPDF2==3.0.1

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from PyPDF2 import PdfReader
import io

@udf(returnType=StringType())
def extract_pdf_text(content):
    if content is None:
        return None
    try:
        reader = PdfReader(io.BytesIO(content))
        return "\n".join(page.extract_text() or "" for page in reader.pages)
    except Exception as e:
        return f"Error: {str(e)}"

df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")

result_df = df.withColumn("text_content", extract_pdf_text("content"))
display(result_df.select("path", "text_content"))

Étape 4.2 : Analyser des images

SQL

SELECT
  path,
  ai_query(
    'databricks-llama-4-maverick',
    'Describe this image in one sentence:',
    files => content
  ) AS description
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.{jpg,jpeg,png}'
)
WHERE _metadata.file_size < 5000000;

Python

result_df = spark.sql("""
    SELECT
      path,
      ai_query(
        'databricks-llama-4-maverick',
        'Describe this image in one sentence:',
        files => content
      ) AS description
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.{jpg,jpeg,png}'
    )
    WHERE _metadata.file_size < 5000000
""")
display(result_df)
Alternative : Extraire les métadonnées d’image sans fonctions IA

Pour extraire les métadonnées d’image sans fonctions IA :

%pip install pillow==10.4.0

from pyspark.sql.functions import udf
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from PIL import Image
import io

image_schema = StructType([
    StructField("width", IntegerType()),
    StructField("height", IntegerType()),
    StructField("format", StringType())
])

@udf(returnType=image_schema)
def get_image_info(content):
    if content is None:
        return None
    try:
        img = Image.open(io.BytesIO(content))
        return {"width": img.width, "height": img.height, "format": img.format}
    except:
        return None

df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.{jpg,jpeg,png}") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")

result_df = df.withColumn("image_info", get_image_info("content"))
display(result_df.select("path", "image_info.*"))

Étape 4.3 : Filtrer et analyser par nom de fichier

Cet exemple filtre les fichiers image avec la sous-chaîne « rose » dans leur nom de fichier.

SQL

SELECT
  path AS file_path,
  ai_query(
    'databricks-llama-4-maverick',
    'Describe this image in one sentence:',
    files => content
  ) AS description
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.{jpg,jpeg,png}'
)
WHERE _metadata.file_name ILIKE '%rose%';

Python

result_df = spark.sql("""
    SELECT
      path AS file_path,
      ai_query(
        'databricks-llama-4-maverick',
        'Describe this image in one sentence:',
        files => content
      ) AS description
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.{jpg,jpeg,png}'
    )
    WHERE _metadata.file_name ILIKE '%rose%'
""")
display(result_df)

Étape 4.4 : Joindre des fichiers avec des tables structurées

Cet exemple utilise des numéros de ligne pour associer des fichiers avec des trajets de taxi à des fins de démonstration. En production, effectuez la jointure sur des clés métier pertinentes.

SQL

-- This example demonstrates joining file metadata with structured data
-- by pairing files with taxi trips using row numbers
WITH files_with_row AS (
  SELECT
    path,
    SPLIT(path, '/')[SIZE(SPLIT(path, '/')) - 1] AS file_name,
    length,
    ROW_NUMBER() OVER (ORDER BY path) AS file_row
  FROM read_files(
    '/Volumes/unstructured_data_lab/raw/files_volume/',
    format => 'binaryFile'
  )
),
trips_with_row AS (
  SELECT
    tpep_pickup_datetime,
    pickup_zip,
    dropoff_zip,
    fare_amount,
    ROW_NUMBER() OVER (ORDER BY tpep_pickup_datetime) AS trip_row
  FROM samples.nyctaxi.trips
  WHERE pickup_zip IS NOT NULL
  LIMIT 5
)
SELECT
  f.path,
  f.file_name,
  f.length,
  t.pickup_zip,
  t.dropoff_zip,
  t.fare_amount,
  t.tpep_pickup_datetime
FROM files_with_row f
INNER JOIN trips_with_row t ON f.file_row = t.trip_row;

Python

from pyspark.sql.functions import col, row_number, element_at, split
from pyspark.sql.window import Window

# Read files and add row numbers
files_df = spark.read.format("binaryFile") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/") \
    .withColumn("file_name", element_at(split(col("path"), "/"), -1))

files_with_row = files_df.alias("files") \
    .withColumn("file_row", row_number().over(Window.orderBy("path")))

# Get trips and add row numbers
trips_df = spark.table("samples.nyctaxi.trips") \
    .filter(col("pickup_zip").isNotNull()) \
    .limit(5)

trips_with_row = trips_df.alias("trips") \
    .withColumn("trip_row", row_number().over(Window.orderBy("tpep_pickup_datetime")))

# Join on row numbers
result_df = files_with_row \
    .join(trips_with_row, col("file_row") == col("trip_row"), "inner") \
    .select(
        "files.path",
        "files.file_name",
        "files.length",
        "trips.pickup_zip",
        "trips.dropoff_zip",
        "trips.fare_amount",
        "trips.tpep_pickup_datetime"
    )

display(result_df)

Étape 5 : Appliquer le contrôle d’accès

Contrôler qui peut lire et écrire des fichiers dans vos volumes. Pour en savoir plus sur la gestion des privilèges dans le catalogue Unity, consultez Gérer les privilèges dans le catalogue Unity.

Étape 5.1 : Accorder l’accès

SQL

-- Replace <user-or-group-name> with your workspace group or user name

-- Grant read access
GRANT READ VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

-- Grant read and write access
GRANT READ VOLUME, WRITE VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

-- Grant all privileges
GRANT ALL PRIVILEGES ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

Python

# Replace <user-or-group-name> with your workspace group or user name
spark.sql("""
    GRANT READ VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

spark.sql("""
    GRANT READ VOLUME, WRITE VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

spark.sql("""
    GRANT ALL PRIVILEGES ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

Explorateur de catalogues

  1. Accédez à l’onglet Autorisations de la page du volume.
  2. Cliquez sur Accorder.
  3. Entrez l’adresse e-mail d’un utilisateur ou le nom d’un groupe.
  4. Sélectionnez les autorisations à accorder.
  5. Cliquez sur Confirmer.

Étape 5.2 : Afficher les privilèges actuels

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Python

display(spark.sql("SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume"))

Explorateur de catalogues

L’onglet Autorisations de la page de volume indique quels utilisateurs et groupes ont accès au volume.

Étape 6 : Configurer l’ingestion incrémentielle

Utilisez le chargeur automatique pour traiter automatiquement les nouveaux fichiers à mesure qu’ils arrivent dans votre volume. Ce modèle est utile pour les flux de travail d’ingestion de données continus. Pour plus d’informations sur les modèles d’ingestion, consultez Modèles de chargement de données courants.

Étape 6.1 : Créer une table de diffusion en continu

SQL

CREATE OR REFRESH STREAMING TABLE document_ingestion
SCHEDULE EVERY 1 HOUR
AS SELECT
  path,
  modificationTime,
  length,
  content,
  _metadata,
  current_timestamp() AS ingestion_time
FROM STREAM(read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/incoming/',
  format => 'binaryFile'
));

Python

from pyspark.sql.functions import current_timestamp, col

dbutils.fs.mkdirs("/Volumes/unstructured_data_lab/raw/files_volume/incoming/")

df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/incoming/")

df_enriched = df \
    .withColumn("ingestion_time", current_timestamp()) \
    .withColumn("source_file", col("_metadata.file_path"))

query = df_enriched.writeStream \
    .option("checkpointLocation",
            "/Volumes/unstructured_data_lab/raw/files_volume/_checkpoints/docs") \
    .trigger(availableNow=True) \
    .toTable("document_ingestion")

query.awaitTermination()

Étape 7 : Partager des fichiers avec OpenSharing

Partagez des volumes en toute sécurité avec des utilisateurs d’autres organisations à l’aide d’OpenSharing. Vous devez créer un destinataire avant le partage. Un destinataire représente une organisation ou un utilisateur externe qui peut accéder à vos données partagées. Consultez Créer des destinataires de données pour OpenSharing (partage Databricks-to-Databricks) pour la configuration des destinataires.

Étape 7.1 : Créer et configurer un partage

SQL

-- Create a share
CREATE SHARE IF NOT EXISTS unstructured_data_share
COMMENT 'Document files for partners';

-- Add the volume
ALTER SHARE unstructured_data_share
ADD VOLUME unstructured_data_lab.raw.files_volume;

-- Create a recipient
CREATE RECIPIENT IF NOT EXISTS <partner_org>
USING ID '<recipient-sharing-identifier>';

-- Grant access
GRANT SELECT ON SHARE unstructured_data_share
TO RECIPIENT <partner_org>;

Python

spark.sql("""
    CREATE SHARE IF NOT EXISTS unstructured_data_share
    COMMENT 'Document files for partners'
""")

spark.sql("""
    ALTER SHARE unstructured_data_share
    ADD VOLUME unstructured_data_lab.raw.files_volume
""")

spark.sql("""
    CREATE RECIPIENT IF NOT EXISTS <partner_org>
    USING ID '<recipient-sharing-identifier>'
""")

spark.sql("""
    GRANT SELECT ON SHARE unstructured_data_share
    TO RECIPIENT <partner_org>
""")

Étape 7.2 : Accéder aux données partagées (en tant que destinataire)

SQL

-- View available shares
SHOW SHARES IN PROVIDER <provider_name>;

-- Create a catalog from the share
CREATE CATALOG IF NOT EXISTS shared_documents
FROM SHARE <provider_name>.unstructured_data_share;

-- Query shared files
SELECT * EXCEPT (content), _metadata
FROM read_files(
  '/Volumes/shared_documents/raw/files_volume/',
  format => 'binaryFile'
)
LIMIT 10;

Python

spark.sql("SHOW SHARES IN PROVIDER <provider_name>").show()

spark.sql("""
    CREATE CATALOG IF NOT EXISTS shared_documents
    FROM SHARE <provider_name>.unstructured_data_share
""")

df = spark.read.format("binaryFile") \
    .load("/Volumes/shared_documents/raw/files_volume/")

df.select("path", "modificationTime", "length").show(10)

Étape 8 : Nettoyer les fichiers

Supprimez les fichiers lorsqu’ils ne sont plus nécessaires.

Python

# Delete a single file
dbutils.fs.rm("/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf")

# Delete a directory recursively
dbutils.fs.rm("/Volumes/unstructured_data_lab/raw/files_volume/sample_files/", recurse=True)

Interface de ligne de commande (CLI)

# Delete a single file
databricks fs rm dbfs:/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf

# Delete a directory recursively
databricks fs rm -r dbfs:/Volumes/unstructured_data_lab/raw/files_volume/sample_files/
Alternative : Utiliser Python standard 
import os
os.remove("/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf")

import shutil
shutil.rmtree("/Volumes/unstructured_data_lab/raw/files_volume/sample_files/")

Ressources supplémentaires

Continuer à en savoir plus sur les volumes

Références de fonction SQL

  • Last updated on