Identifier et supprimer des points de terminaison de recherche vectorielle vides

Cette page décrit comment identifier et supprimer des points de terminaison de recherche vectorielle vides. Étant donné que les points de terminaison de recherche vectorielle sont des ressources spécifiques à l’espace de travail, vous devez répéter ce processus pour chaque espace de travail.

Spécifications

Kit de développement logiciel (SDK) Databricks pour Python (databricks-sdk).
Kit de développement logiciel (SDK) Python de recherche vectorielle Databricks (databricks-vectorsearch).
Authentification configurée (profils OAuth, PAT ou configuration).
CAN_MANAGE autorisation d'accès aux points d'accès de recherche vectorielle dans les espaces de travail cibles.

Pour installer les kits SDK requis dans votre notebook Databricks ou votre environnement Python local :

# In a Databricks notebook
%pip install databricks-sdk databricks-vectorsearch

# In local Python environment
# pip install databricks-sdk databricks-vectorsearch

Identifier les points de terminaison vides

Dans l’interface utilisateur Databricks, les points de terminaison de recherche vectorielle sont affichés sous l’onglet Recherche vectorielle de l’écran Calcul . Cochez la case Points de terminaison vides pour afficher les points de terminaison qui n’ont aucun index associé à eux. Les points de terminaison vides sont également marqués avec une icône de triangle d’avertissement, comme indiqué.

Authentication

Cette section décrit les options d’authentification.

Option numéro 1. Exécuter dans un notebook Databricks

Lorsque vous exécutez le code dans un notebook d’espace de travail Databricks, l’authentification est automatique :

from databricks.vector_search.client import VectorSearchClient

# Credentials are picked up automatically from notebook context
client = VectorSearchClient()

Option 2. Jeton d’accès personnel (PAT)

Pour les environnements externes, fournissez des informations d’identification explicites :

from databricks.vector_search.client import VectorSearchClient

client = VectorSearchClient(
    workspace_url="https://<your-instance>.cloud.databricks.com",
    personal_access_token="dapiXXXXXXXXXXXXXXXXXXXXXXXX"
)

Option 3. Utiliser des profils de configuration (recommandés pour plusieurs espaces de travail)

Créez un .databrickscfg fichier dans votre répertoire de base et incluez un profil pour chaque espace de travail :

[DEFAULT]
host = https://workspace1.cloud.databricks.com
token = dapiXXXXXXXXXXXXXXXXXXXXXXXX

[PRODUCTION]
host = https://workspace2.cloud.databricks.com
token = dapiYYYYYYYYYYYYYYYYYYYYYYYY

[DEVELOPMENT]
host = https://workspace3.cloud.databricks.com
token = dapiZZZZZZZZZZZZZZZZZZZZZZZZ

Si vous préférez ne pas utiliser de profils de configuration, vous pouvez spécifier des informations d’identification directement :

# Define workspaces with explicit credentials
workspace_configs = [
    {
        'workspace_url': 'https://workspace1.cloud.databricks.com',
        'token': 'dapiXXXXXXXXXXXXXXXXXXXXXXXX'
    },
    {
        'workspace_url': 'https://workspace2.cloud.databricks.com',
        'token': 'dapiYYYYYYYYYYYYYYYYYYYYYYYY'
    }
]

# Run cleanup, set `dry_run=False` to perform actual deletion
results = cleanup_multiple_workspaces(workspace_configs, dry_run=True)

Supprimer des endpoints dans un espace de travail

Les points de terminaison de recherche vectorielle sont spécifiques à l’espace de travail. Voici un script de base pour rechercher et supprimer des points de terminaison vides dans un seul espace de travail. Pour nettoyer les points de terminaison vides entre plusieurs espaces de travail, consultez Supprimer des points de terminaison sur plusieurs espaces de travail.

Important

La suppression du point de terminaison est irréversible. Utilisez l’option dry_run=True pour afficher la liste des points de terminaison qui seront supprimés. Une fois que vous avez confirmé que la liste est correcte, exécutez le script avec dry_run=False.

from databricks.vector_search.client import VectorSearchClient

def cleanup_empty_endpoints(client, dry_run=True):
    """
    Find and delete empty Vector Search endpoints.

    Args:
        client: VectorSearchClient instance
        dry_run: If True, only print what would be deleted without actually deleting

    Returns:
        List of deleted endpoint names
    """
    deleted_endpoints = []

    # List all Vector Search endpoints
    endpoints = client.list_endpoints()

    for endpoint in endpoints["endpoints"]:
        # List indexes in this endpoint
        indexes = list(client.list_indexes(name=endpoint["name"])['vector_indexes'])

        if len(indexes) == 0:
            if dry_run:
                print(f"[DRY RUN] Would delete empty endpoint: '{endpoint["name"]}'")
            else:
                print(f"Deleting empty endpoint: '{endpoint["name"]}'")
                try:
                    client.delete_endpoint(endpoint["name"])
                    deleted_endpoints.append(endpoint["name"])
                    print(f"✓ Successfully deleted: {endpoint["name"]}")
                except Exception as e:
                    print(f"✗ Failed to delete {endpoint["name"]}: {str(e)}")
        else:
            print(f"Endpoint '{endpoint["name"]}' has {len(indexes)} indexes - keeping")

    return deleted_endpoints

# Example usage
client = VectorSearchClient()  # Uses default authentication
# Set `dry_run=False` when you are ready to delete endpoints
deleted = cleanup_empty_endpoints(client, dry_run=True)
print(f"\nTotal endpoints deleted: {len(deleted)}")

Supprimer des points de terminaison sur plusieurs espaces de travail

Pour nettoyer les points de terminaison vides entre plusieurs espaces de travail, effectuez une itération dans vos profils de configuration :

Important

La suppression du point de terminaison est irréversible. Utilisez l’option dry_run=True pour afficher la liste des points de terminaison qui seront supprimés. Une fois que vous avez confirmé que la liste est correcte, exécutez le script avec dry_run=False.

Lors du traitement de nombreux espaces de travail, tenez compte des limites de débit d’API. Ajoutez des délais si nécessaire.

import time

for config in workspace_configs:
    # Set `dry_run=False` to perform actual deletion
    result = cleanup_workspace(**config, dry_run=True)
    time.sleep(2)  # Add delay between workspaces

from databricks.sdk import WorkspaceClient
from databricks.vector_search.client import VectorSearchClient
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def cleanup_workspace(profile_name=None, workspace_url=None, token=None, dry_run=True):
    """
    Clean up empty endpoints in a specific workspace.

    Args:
        profile_name: Name of configuration profile to use
        workspace_url: Direct workspace URL (if not using profile)
        token: PAT token (if not using profile)
        dry_run: If True, only show what would be deleted

    Returns:
        Dict with cleanup results
    """
    try:
        # Initialize client based on authentication method
        if profile_name:
            # Use Databricks SDK to get credentials from profile
            w = WorkspaceClient(profile=profile_name)
            workspace_url = w.config.host
            client = VectorSearchClient(
                workspace_url=workspace_url,
                personal_access_token=w.config.token
            )
            logger.info(f"Connected to workspace using profile '{profile_name}': {workspace_url}")
        elif workspace_url and token:
            client = VectorSearchClient(
                workspace_url=workspace_url,
                personal_access_token=token
            )
            logger.info(f"Connected to workspace: {workspace_url}")
        else:
            # Use default authentication (notebook context)
            client = VectorSearchClient()
            logger.info("Connected using default authentication")

        # Perform cleanup
        deleted = cleanup_empty_endpoints(client, dry_run=dry_run)

        return {
            'workspace': workspace_url or 'default',
            'success': True,
            'deleted_count': len(deleted),
            'deleted_endpoints': deleted
        }

    except Exception as e:
        logger.error(f"Failed to process workspace: {str(e)}")
        return {
            'workspace': workspace_url or profile_name or 'default',
            'success': False,
            'error': str(e)
        }

def cleanup_multiple_workspaces(workspace_configs, dry_run=True):
    """
    Clean up empty endpoints across multiple workspaces.

    Args:
        workspace_configs: List of workspace configurations
        dry_run: If True, only show what would be deleted

    Returns:
        Summary of cleanup results
    """
    results = []

    for config in workspace_configs:
        logger.info(f"\n{'='*60}")
        result = cleanup_workspace(**config, dry_run=dry_run)
        results.append(result)
        logger.info(f"{'='*60}\n")

    # Print summary
    total_deleted = sum(r['deleted_count'] for r in results if r['success'])
    successful = sum(1 for r in results if r['success'])
    failed = sum(1 for r in results if not r['success'])

    logger.info("\n" + "="*60)
    logger.info("CLEANUP SUMMARY")
    logger.info("="*60)
    logger.info(f"Workspaces processed: {len(results)}")
    logger.info(f"Successful: {successful}")
    logger.info(f"Failed: {failed}")
    logger.info(f"Total endpoints deleted: {total_deleted}")

    if failed > 0:
        logger.warning("\nFailed workspaces:")
        for r in results:
            if not r['success']:
                logger.warning(f"  - {r['workspace']}: {r['error']}")

    return results

# Example: Clean up using configuration profiles
workspace_configs = [
    {'profile_name': 'DEFAULT'},
    {'profile_name': 'PRODUCTION'},
    {'profile_name': 'DEVELOPMENT'}
]

# Set `dry_run=False` to do actual deletion.
results = cleanup_multiple_workspaces(workspace_configs, dry_run=True)

Filtrage personnalisé

Vous pouvez ajouter une logique personnalisée pour exclure certains points de terminaison de la suppression, comme indiqué :

def should_delete_endpoint(endpoint, indexes):
    """
    Custom logic to determine if an endpoint should be deleted.

    Args:
        endpoint: Endpoint object
        indexes: List of indexes in the endpoint

    Returns:
        Boolean indicating if endpoint should be deleted
    """
    # Don't delete if it has indexes
    if len(indexes) > 0:
        return False

    # Don't delete endpoints with specific naming patterns
    protected_patterns = ['prod-', 'critical-', 'do-not-delete']
    for pattern in protected_patterns:
        if pattern in endpoint.name.lower():
            logger.warning(f"Skipping protected endpoint: {endpoint.name}")
            return False

    # Add more custom logic as needed
    return True

Exporter les résultats

Pour enregistrer les résultats de nettoyage dans un fichier pour l’audit :

import json
from datetime import datetime

def export_results(results, filename=None):
    """Export cleanup results to JSON file."""
    if not filename:
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = f'vector_search_cleanup_{timestamp}.json'

    with open(filename, 'w') as f:
        json.dump({
            'timestamp': datetime.now().isoformat(),
            'results': results
        }, f, indent=2)

    logger.info(f"Results exported to: {filename}")

Résolution des problèmes

Problèmes d’authentification

Vérifiez que vos jetons PAT sont valides et non expirés.
Vérifiez que les profils de configuration sont correctement mis en forme.
Vérifiez que vos jetons disposent des autorisations nécessaires.

Erreurs d’autorisation

Vérifiez que votre utilisateur ou principal de service dispose CAN_MANAGE de l’autorisation nécessaire pour les points de terminaison de recherche vectorielle.

Problèmes réseau

Pour les environnements avec des exigences de proxy, configurez le Kit de développement logiciel (SDK) de manière appropriée :

import os
os.environ['HTTPS_PROXY'] = 'http://your-proxy:po

Étapes suivantes

Planifiez l’exécution périodique de ce script à l’aide des Travaux Lakeflow.
Intégrez-vous à votre pipeline d'infrastructure en tant que code.
Ajoutez des notifications par e-mail ou Slack pour les résumés de nettoyage.
Créez un tableau de bord pour suivre l’utilisation des terminaux dans les espaces de travail.

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2025-11-15