Identificar e excluir pontos de extremidade vazios da Pesquisa de Vetor

Esta página descreve como você pode identificar e excluir pontos de extremidade vazios da Pesquisa de Vetores. Como os pontos de extremidade da Pesquisa de Vector são recursos específicos do workspace, você precisa repetir esse processo para cada workspace separadamente.

Requirements

SDK do Databricks para Python (databricks-sdk).
SDK do Python de Pesquisa de Vetor do Databricks (databricks-vectorsearch).
Autenticação configurada (OAuth, PAT ou perfis de configuração).
CAN_MANAGE permissão para pontos de extremidade de Pesquisa vetor em workspaces de destino.

Para instalar os SDKs necessários no notebook do Databricks ou no ambiente local do Python:

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

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

Identificar pontos de extremidade vazios

Na interface do usuário do Databricks, os pontos de extremidade de pesquisa de vetor são mostrados na guia Pesquisa de Vetor da tela de computação . Alterne a caixa de seleção Pontos de Extremidade Vazios para exibir pontos de extremidade que não têm índices associados a eles. Pontos de extremidade vazios também são marcados com um ícone de triângulo de aviso, conforme mostrado.

Authentication

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

Opção 1. Executar dentro de um notebook do Databricks

Quando você executa o código em um notebook de workspace do Databricks, a autenticação é automática:

from databricks.vector_search.client import VectorSearchClient

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

Opção 2. PAT (token de acesso pessoal)

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

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

Crie um .databrickscfg arquivo em seu diretório inicial 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ê poderá especificar credenciais diretamente:

# 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 pontos de extremidade em um único workspace

Os pontos de extremidade de Pesquisa de Vector são específicos do workspace. Aqui está um script básico para localizar e excluir pontos de extremidade vazios em um único workspace. Para limpar pontos de extremidade vazios em vários workspaces, consulte Excluir pontos de extremidade em vários workspaces.

Importante

A exclusão de ponto de extremidade é irreversível. Use a opção dry_run=True para ver uma lista dos pontos de extremidade que serão excluídos. Depois de confirmar que a lista está correta, execute o script com 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)}")

Excluir pontos de extremidade em vários workspaces

Para limpar pontos de extremidade vazios em vários workspaces, iterar por meio de seus perfis de configuração:

Importante

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

Ao processar muitos workspaces, esteja atento aos limites de taxa de API. Adicione atrasos, se necessário:

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)

Filtragem personalizada

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

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 auditoria:

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

Resolução de problemas

Problemas de autenticação

Verifique se os tokens PAT são válidos e não expiraram.
Verifique se os perfis de configuração estão formatados corretamente.
Verifique se os tokens têm as permissões necessárias.

Erros de permissão

Verifique se o usuário ou a entidade de serviço tem CAN_MANAGE permissão para pontos de extremidade de Pesquisa de Vetor.

Problemas de rede

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

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

Próximas etapas

Agende esse script para ser executado periodicamente usando Trabalhos do Lakeflow.
Integre-se ao pipeline de infraestrutura como código.
Adicione notificações de email ou slack para resumos de limpeza.
Crie um painel para acompanhar o uso do ponto de extremidade entre workspaces.

Comentários

Esta página foi útil?

Last updated on 2025-11-15