Vector Retrieval Service for Milvus:Migrar dados para o Alibaba Cloud Milvus com uma ferramenta de imagem

Requisitos de rede

O contêiner precisa de acesso de rede às instâncias de origem e de destino do Alibaba Cloud Milvus. Para melhor desempenho, implante o contêiner na mesma VPC da instância de destino.

docker pull registry.cn-hangzhou.aliyuncs.com/taihao-executor/taihao-executor:release_2.22.0-ali

1. Inicie o contêiner em modo desanexado.

   docker run -d -it \
     --name milvus-migration \
     registry.cn-hangzhou.aliyuncs.com/taihao-executor/taihao-executor:release_2.22.0-ali \
     /bin/bash

2. Localize o ID do contêiner e acesse-o.

   # Find the container
   docker ps
   
   # Enter the container (replace with your actual container ID)
   docker exec -it <container_id> bash

   Exemplo:

   docker exec -it 55ac98f3b054 bash

vi migration.conf

Modelo de configuração

env {
  parallelism = 1           # Concurrency level. We recommend setting this to 1 initially.
  job.mode = "BATCH"        # Batch processing mode.
}

source {
  Milvus {
    url = "http://<source-instance-endpoint>:19530"       # An internal endpoint is supported.
    token = "<username>:<password>"                 # Example: root:Test123456@
    database = "default"                    # The database to migrate from. Defaults to `default`.
    collections = ["col_a", "col_b"]        # A list of collections to migrate.
    batch_size = 10000                      # Entities to read per batch. Increase this for large collections.
  }
}

sink {
  Milvus {
    url = "http://<target-Alibaba-Cloud-Milvus-endpoint>:19530"
    token = "<target-instance-token>"
    database = "default"
    batch_size = 1000
    enable_auto_id = false                 # Set to false to preserve auto-generated IDs from the source. Otherwise, set to true.
  }
}

Observações

• Para evitar falhas, carregue todas as coleções a migrar na memória com o método load().

• Para migrar todas as coleções carregadas, omita o parâmetro collections do arquivo de configuração.

• Se o contêiner e a instância de destino estiverem na mesma região, use um endpoint interno para acelerar a transferência.

Opção 1: Executar em modo local (máquina única)

nohup ./bin/seatunnel.sh --config ./migration.conf -m local > migration.log 2>&1 &

-Xms4g
-Xmx8g

Opção 2: Executar em modo cluster (recomendado para alto desempenho)

Use este modo para migrar grandes volumes de dados.

# Create a log directory
mkdir -p ./logs

# Start the cluster service
./bin/seatunnel-cluster.sh -d

# Submit the task
nohup ./bin/seatunnel.sh --config ./migration.conf > migration.log 2>&1 &

1. Crie um índice.

   milvus_client = milvus.prepare_index_params()
   index_params.add_index(
           field_name="vector",  # Name of the vector field to be indexed
           index_type="HNSW",  # Type of the index to create
           index_name="vector_index",  # Name of the index to create
           metric_type="L2",  # Metric type used to measure similarity
           params={
               "M": 64,  # Maximum number of neighbors each node can connect to in the graph
               "efConstruction": 100  # Number of candidate neighbors considered for connection during index construction
           }  # Index building params
       )
   milvus_client.create_index("collectionName", index_params)

2. Carregue a coleção na memória.

   milvus_client.load_collection()

   Crie um índice antes de carregar a coleção para habilitar a busca acelerada. Os principais parâmetros são:

   Parâmetro	Como obter
   url	Acesse o console do Alibaba Cloud Milvus. Na aba Security Configuration, visualize o endpoint público ou interno. Prefira o endpoint interno para melhor desempenho.
   token	O formato é username:password (ex.: root:YourPassword123@). No console do Alibaba Cloud Milvus, abra a aba Security Configuration e consulte a senha da conta root.
   database	O valor padrão é default. Se usar o recurso de múltiplos bancos de dados, obtenha outros nomes chamando a API list_databases().

   Exemplo completo de configuração:

   env {
     parallelism = 1
     job.mode = "BATCH"
   }
   
   source {
     Milvus {
       url = "http://xx.xx.xx.xx:19530"
       token = "root:SourcePass123@"
       database = "default"
       collections = ["medium_articles"]
       batch_size = 10000
     }
   }
   
   sink {
     Milvus {
       url = "http://proxy-bj.vpc.milvus.aliyuncs.com:19530"
       token = "root:TargetPass123@"
       database = "default"
       batch_size = 10000
       enable_auto_id = false
     }
   }

P1: Por que recebo o erro "Collection not loaded" durante a migração?

R: Verifique se todas as coleções de origem foram carregadas na memória com o método .load().

P2: É possível migrar apenas campos específicos?

R: Não. A versão atual suporta apenas a migração de coleções inteiras. A filtragem de campos específicos não é suportada.

P3: Como monitorar o progresso da migração?

R: Acompanhe a migração verificando a saída no arquivo migration.log ou usando o Attu para observar a contagem de linhas na coleção de destino.