Identificar e excluir endpoints de pesquisa vetorial vazios

Esta página descreve como você pode identificar e excluir um ponto de extremidade de pesquisa vetorial vazio. Como os endpoints do Vector Search são recursos específicos workspace , você precisa repetir esse processo para cada workspace separadamente.

Requisitos

SDK do Databricks para Python (databricks-sdk).
SDK Python de pesquisa vetorial do Databricks (databricks-vectorsearch).
Autenticação configurada (OAuth, PAT ou perfis de configuração).
CAN_MANAGE Permissão para o endpoint de pesquisa vetorial no espaço de trabalho de destino.

Para instalar os SDKs necessários no seu Notebook Databricks ou no seu ambiente Python local:

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

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

Identificar ponto final vazio

Na interface Databricks , os pontos de extremidade de pesquisa vetorial são exibidos na tab "Pesquisa Vetorial" da tela de computação . Marque a caixa de seleção "Ponto de extremidade vazio" para exibir os pontos de extremidade que não possuem índices associados. Os pontos finais vazios também são marcados com um ícone de triângulo de aviso, conforme mostrado.

Autenticação

Esta seção descreve as opções de autenticação.

Opção 1. execução dentro de um notebook Databricks

Ao executar o código em um Notebook Databricks workspace , a autenticação é automática:

Python
from databricks.vector_search.client import VectorSearchClient

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

Opção 2. access token pessoal (PAT)

Para ambientes externos, forneça credenciais explícitas:

Python
from databricks.vector_search.client import VectorSearchClient

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

Opção 3. Usar perfis de configuração (Recomendado para vários espaços de trabalho)

Crie um arquivo .databrickscfg em seu diretório pessoal e inclua um perfil para cada workspace:

[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

Se preferir não usar perfis de configuração, você pode especificar as credenciais diretamente:

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

Excluir ponto de extremidade em um único workspace

Os pontos de extremidade da pesquisa vetorial são específicos workspace . Aqui está um script básico para encontrar e excluir endpoints vazios em um único workspace. Para limpar endpoints vazios em vários espaços de trabalho, consulte Excluir endpoint em vários espaços de trabalho.

importante

A exclusão do endpoint é irreversível. Use a opção dry_run=True para ver uma lista dos endpoints que serão excluídos. Depois de confirmar que a lista está correta, execute o script com dry_run=False.

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

Excluir ponto de extremidade em vários espaços de trabalho

Para limpar endpoints vazios em vários espaços de trabalho, percorra seus perfis de configuração:

importante

A exclusão do endpoint é irreversível. Use a opção dry_run=True para ver uma lista dos endpoints que serão excluídos. Depois de confirmar que a lista está correta, execute o script com dry_run=False.

Ao processar muitos espaços de trabalho, fique atento aos limites de taxa API . Adicione atrasos, se necessário:

Python
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

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

Filtragem personalizada

Você pode adicionar lógica personalizada para excluir determinados endpoints da exclusão, conforme mostrado:

Python
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

Exportar resultados

Para salvar os resultados da limpeza em um arquivo para fins de auditoria:

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

Solução de problemas

Problemas de autenticação

Verifique se seus tokens PAT são válidos e não expiraram.
Certifique-se de que os perfis de configuração estejam formatados corretamente.
Verifique se seus tokens possuem as permissões necessárias.

Erros de permissão

Verifique se o seu usuário ou entidade de serviço tem permissão CAN_MANAGE para o endpoint de pesquisa vetorial.

Problemas de rede

Para ambientes com requisitos de proxy, configure o SDK adequadamente:

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

Próximos passos

programar este script para execução periodicamente usando LakeFlow Jobs.
Integre com seu pipeline Infrastructure-as-Code .
Adicione notificações email ou Slack para receber resumos da limpeza.
Crie um painel de controle para monitorar o uso endpoint em todo o espaço de trabalho.

Requisitos​

Identificar ponto final vazio​

Autenticação​

Opção 1. execução dentro de um notebook Databricks​

Opção 2. access token pessoal (PAT)​

Opção 3. Usar perfis de configuração (Recomendado para vários espaços de trabalho)​

Excluir ponto de extremidade em um único workspace​

Excluir ponto de extremidade em vários espaços de trabalho​

Filtragem personalizada​

Exportar resultados​

Solução de problemas​

Problemas de autenticação​

Erros de permissão​

Problemas de rede​

Próximos passos​