Identificar e excluir pontos de extremidade vazios da Pesquisa Vetorial

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

Requerimentos

Databricks SDK para Python (databricks-sdk).
Databricks Vetor Search Python SDK (databricks-vectorsearch).
Autenticação configurada (OAuth, PAT ou perfis de configuração).
CAN_MANAGE permissão para pontos de extremidade de Pesquisa Vetorial em espaços de trabalho de destino.

Para instalar os SDKs necessários em seu notebook Databricks ou ambiente Python local:

# 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 vetorial são mostrados na guia Pesquisa vetorial da tela 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. Os pontos finais 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 bloco de anotações Databricks

Quando você executa o código em um bloco de anotações do espaço de trabalho 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. Token de acesso pessoal (PAT)

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 espaços de trabalho)

Crie um .databrickscfg arquivo em seu diretório base e inclua um perfil para cada espaço de trabalho:

[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 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 espaço de trabalho

Os pontos de extremidade da Pesquisa Vetorial são específicos do espaço de trabalho. Aqui está um script básico para localizar e excluir pontos de extremidade vazios em um único espaço de trabalho. Para limpar pontos de extremidade vazios em vários espaços de trabalho, consulte Excluir pontos de extremidade em vários espaços de trabalho.

Importante

A eliminação do parâmetro de avaliação é 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 espaços de trabalho

Para limpar pontos de extremidade vazios em vários espaços de trabalho, itere através de seus perfis de configuração:

Importante

A eliminação do parâmetro de avaliação é 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 espaços de trabalho, esteja atento aos limites de taxa de API. Adicionar 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}")

Solução de problemas

Problemas de autenticação

Verifique se os seus tokens PAT são válidos e não expiraram.
Verifique se os perfis de configuração estão formatados corretamente.
Verifique se seus 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 Vetorial.

Problemas de rede

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

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

Próximos passos

Agende este script para ser executado periodicamente usando o Lakeflow Jobs.
Integre com seu pipeline de infraestrutura como código.
Adicione notificações por e-mail ou do Slack para resumos de limpeza.
Crie um painel para acompanhar o uso do ponto de extremidade entre espaços de trabalho.

Feedback

Esta página foi útil?

Last updated on 2025-11-15