Identifizieren und Löschen leerer Vektorsuchendpunkte

Auf dieser Seite wird beschrieben, wie Sie leere Vektorsuchendpunkte identifizieren und löschen können. Da Vektorsuchendpunkte arbeitsbereichspezifische Ressourcen sind, müssen Sie diesen Vorgang für jeden Arbeitsbereich separat wiederholen.

Anforderungen

• Databricks SDK für Python (databricks-sdk).
• Databricks Vector Search Python SDK (databricks-vectorsearch).
• Authentifizierung konfiguriert (OAuth, PAT oder Konfigurationsprofile).
• CAN_MANAGE Berechtigung für Vektorsuchendpunkte in Zielarbeitsbereichen.

So installieren Sie die erforderlichen SDKs in Ihrem Databricks-Notizbuch oder ihrer lokalen Python-Umgebung:

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

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

Identifizieren leerer Endpunkte

Auf der Benutzeroberfläche von Databricks werden Vektorsuchendpunkte auf der Registerkarte " Vektorsuche " des Bildschirms " Berechnen " angezeigt. Aktivieren Sie das Kontrollkästchen "Leere Endpunkte ", um Endpunkte anzuzeigen, denen keine Indizes zugeordnet sind. Leere Endpunkte werden auch mit einem Warndreiecksymbol gekennzeichnet, wie dargestellt.

Authentifizierung

In diesem Abschnitt werden die Authentifizierungsoptionen beschrieben.

Option 1: Ausführen in einem Databricks-Notizbuch

Wenn Sie den Code in einem Databricks-Arbeitsbereichsnotizbuch ausführen, wird die Authentifizierung automatisch ausgeführt:

from databricks.vector_search.client import VectorSearchClient

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

Option 2. Persönliches Zugriffstoken (Personal Access Token, PAT)

Geben Sie für externe Umgebungen explizite Anmeldeinformationen an:

from databricks.vector_search.client import VectorSearchClient

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

Option 3. Verwenden von Konfigurationsprofilen (empfohlen für mehrere Arbeitsbereiche)

Erstellen Sie eine .databrickscfg Datei in Ihrem Startverzeichnis, und fügen Sie ein Profil für jeden Arbeitsbereich ein:

[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

Wenn Sie konfigurationsprofile nicht verwenden möchten, können Sie Anmeldeinformationen direkt angeben:

# 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)

Löschen von Endpunkten in einem einzelnen Arbeitsbereich

Vektorsuchendpunkte sind arbeitsbereichspezifisch. Hier ist ein einfaches Skript zum Suchen und Löschen leerer Endpunkte in einem einzigen Arbeitsbereich. Informationen zum Bereinigen leerer Endpunkte über mehrere Arbeitsbereiche hinweg finden Sie unter "Löschen von Endpunkten über mehrere Arbeitsbereiche hinweg".

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)}")

Löschen von Endpunkten über mehrere Arbeitsbereiche

Um leere Endpunkte über mehrere Arbeitsbereiche hinweg zu bereinigen, durchlaufen Sie Ihre Konfigurationsprofile:

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)

Benutzerdefinierte Filterung

Sie können benutzerdefinierte Logik hinzufügen, um bestimmte Endpunkte vom Löschen auszuschließen, wie gezeigt:

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

Exportieren von Ergebnissen

So speichern Sie Bereinigungsergebnisse in einer Datei für die Überwachung:

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}")

Problembehandlung

Authentifizierungsprobleme

• Überprüfen Sie, ob Ihre PAT-Token gültig und nicht abgelaufen sind.
• Stellen Sie sicher, dass Konfigurationsprofile korrekt formatiert sind.
• Überprüfen Sie, ob Ihre Token über die erforderlichen Berechtigungen verfügen.

Berechtigungsfehler

Stellen Sie sicher, dass Ihr Benutzer oder Dienstprinzipal über die Berechtigung für Vektorsuchendpunkte verfügt CAN_MANAGE .

Netzwerkprobleme

Konfigurieren Sie für Umgebungen mit Proxyanforderungen das SDK entsprechend:

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

Nächste Schritte

• Planen Sie dieses Skript so, dass es regelmäßig mithilfe von Lakeflow-Aufträgen ausgeführt wird.
• Integration in Ihre Infrastruktur-as-Code-Pipeline.
• Fügen Sie E-Mail- oder Slack-Benachrichtigungen für Bereinigungszusammenfassungen hinzu.
• Erstellen Sie ein Dashboard, um die Endpunktnutzung über Arbeitsbereiche hinweg nachzuverfolgen.