Identifiera och ta bort tomma slutpunkter för vektorsökning

2025-08-21

På den här sidan beskrivs hur du kan identifiera och ta bort tomma slutpunkter för vektorsökning. Eftersom slutpunkter för vektorsökning är arbetsytespecifika resurser måste du upprepa den här processen för varje arbetsyta separat.

Kravspecifikation

Databricks SDK för Python (databricks-sdk).
Databricks Vector Search Python SDK (databricks-vectorsearch).
Autentisering konfigurerad (OAuth-, PAT- eller konfigurationsprofiler).
CAN_MANAGE behörighet för slutpunkter för vektorsökning i målarbetsytor.

Så här installerar du nödvändiga SDK:er i din Databricks-notebook-fil eller lokala Python-miljö:

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

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

Identifiera tomma slutpunkter

I Databricks-användargränssnittet visas slutpunkter för vektorsökning på fliken Vektorsökning på skärmen Beräkning . Växla kryssrutan Tomma slutpunkter för att visa slutpunkter som inte har några index associerade med dem. Tomma slutpunkter markeras också med en varningstriangelikon som visas.

Authentication

I det här avsnittet beskrivs autentiseringsalternativ.

Alternativ 1. Köra i en Databricks-notebook-fil

När du kör koden i en Databricks-arbetsyteanteckningsbok sker autentisering automatiskt:

from databricks.vector_search.client import VectorSearchClient

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

Alternativ 2. Personlig åtkomsttoken (PAT)

För externa miljöer anger du explicita autentiseringsuppgifter:

from databricks.vector_search.client import VectorSearchClient

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

Alternativ 3. Använda konfigurationsprofiler (rekommenderas för flera arbetsytor)

Skapa en .databrickscfg fil i din hemkatalog och inkludera en profil för varje arbetsyta:

[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

Om du föredrar att inte använda konfigurationsprofiler kan du ange autentiseringsuppgifter direkt:

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

Ta bort slutpunkter på en enda arbetsyta

Slutpunkter för vektorsökning är arbetsytespecifika. Här är ett grundläggande skript för att hitta och ta bort tomma slutpunkter på en enda arbetsyta. Information om hur du rensar tomma slutpunkter över flera arbetsytor finns i Ta bort slutpunkter över flera arbetsytor.

Viktigt!

Slutpunktsborttagningen kan inte ångras. Använd alternativet dry_run=True för att se en lista över de slutpunkter som ska tas bort. När du har bekräftat att listan är korrekt kör du skriptet med 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)}")

Ta bort slutpunkter över flera arbetsytor

Om du vill rensa tomma slutpunkter över flera arbetsytor itererar du genom dina konfigurationsprofiler:

Viktigt!

Slutpunktsborttagningen kan inte ångras. Använd alternativet dry_run=True för att se en lista över de slutpunkter som ska tas bort. När du har bekräftat att listan är korrekt kör du skriptet med dry_run=False.

När du bearbetar många arbetsytor bör du tänka på API-hastighetsbegränsningar. Lägg till fördröjningar om det behövs:

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)

Anpassad filtrering

Du kan lägga till anpassad logik för att undanta vissa slutpunkter från borttagning, enligt följande:

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

Exportera resultat

Så här sparar du rensningsresultat i en fil för granskning:

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

Felsökning

Autentiseringsproblem

Kontrollera att dina PAT-token är giltiga och inte har upphört att gälla.
Se till att konfigurationsprofilerna är korrekt formaterade.
Kontrollera att dina token har nödvändiga behörigheter.

Behörighetsfel

Kontrollera att användaren eller tjänstens huvudnamn har CAN_MANAGE behörighet för slutpunkter för vektorsökning.

Nätverksproblem

För miljöer med proxykrav konfigurerar du SDK:n på rätt sätt:

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

Nästa steg

Schemalägg skriptet så att det körs regelbundet med Lakeflow-jobb.
Integrera med din pipeline för infrastruktur som kod.
Lägg till e-post- eller Slack-meddelanden för rensningssammanfattningar.
Skapa en instrumentpanel för att spåra slutpunktsanvändning mellan arbetsytor.

Feedback

Var den här sidan till hjälp?