¿Cómo se procesan los datos en una base de datos vectorial?

Imagen de portada

Este artículo ha sido escrito por Zhenshan Cao y transcrito por Angela Ni.

En las dos entradas anteriores de esta serie de blogs, ya hemos cubierto la arquitectura del sistema de Milvus, la base de datos vectorial más avanzada del mundo, y su SDK y API de Python.

El objetivo principal de esta entrada es ayudarle a comprender cómo se procesan los datos en Milvus, profundizando en el sistema Milvus y examinando la interacción entre los componentes de procesamiento de datos.

A continuación se enumeran algunos recursos útiles antes de empezar. Recomendamos leerlos primero para comprender mejor el tema de este post.

• Profundización en la arquitectura de Milvus
• Modelo de datos Milvus
• El papel y la función de cada componente de Milvus
• Procesamiento de datos en Milvus

Interfaz MsgStream

La interfaz Ms gStream es crucial para el procesamiento de datos en Milvus. Cuando se llama a Start(), la coroutine en segundo plano escribe datos en el log broker o lee datos de allí. Cuando se llama a Close(), la coroutine se detiene.

Interfaz MsgStream

El MsgStream puede servir como productor y consumidor. La interfaz AsProducer(channels []string) define el MsgStream como productor, mientras que la AsConsumer(channels []string, subNamestring)lo define como consumidor. El parámetro channels se comparte en ambas interfaces y se utiliza para definir en qué canales (físicos) se escriben los datos o de cuáles se leen.

El número de shards de una colección puede especificarse cuando se crea una colección. Cada shard corresponde a un canal virtual (vchannel). Por lo tanto, una colección puede tener múltiples vcanales. Milvus asigna a cada canal virtual del agente de registro un canal físico (pchannel).

Cada canal virtual/shard corresponde a un canal físico.

Produce() en la interfaz MsgStream encargada de escribir datos en los pchannels del log broker. Los datos pueden escribirse de dos maneras:

• Escritura única: las entidades se escriben en diferentes shards (vchannel) mediante los valores hash de las claves primarias. A continuación, estas entidades fluyen a los pcanales correspondientes en el log broker.
• Broadcast write: las entidades se escriben en todos los pchannels especificados por el parámetro channels.

Consume() es un tipo de API de bloqueo. Si no hay datos disponibles en el pchannel especificado, la coroutine se bloqueará cuando se llame a Consume() en la interfaz MsgStream. Por otro lado, Chan() es una API no bloqueante, lo que significa que la coroutina lee y procesa datos sólo si hay datos existentes en el pchannel especificado. En caso contrario, la coroutina puede procesar otras tareas y no se bloqueará cuando no haya datos disponibles.

Seek() es un método de recuperación de fallos. Cuando se inicia un nuevo nodo, se puede obtener el registro de consumo de datos y reanudar el consumo de datos desde donde se interrumpió llamando a Seek().

Escritura de datos

Los datos escritos en diferentes canales virtuales (shards) pueden ser mensajes de inserción o mensajes de borrado. Estos vchannels también pueden denominarse DmChannels (canales de manipulación de datos).

Diferentes colecciones pueden compartir los mismos pchannels en el log broker. Una colección puede tener múltiples shards y, por tanto, múltiples vchannels correspondientes. En consecuencia, las entidades de una misma colección fluyen hacia múltiples pchannels correspondientes en el log broker. Como resultado, el beneficio de compartir pchannels es un mayor volumen de rendimiento habilitado por la alta concurrencia del log broker.

Cuando se crea una colección, no sólo se especifica el número de shards, sino que también se decide la asignación entre vchannels y pchannels en el log broker.

Ruta de escritura en Milvus

Como se muestra en la ilustración anterior, en la ruta de escritura, los proxies escriben datos en el log broker a través de la interfaz AsProducer() del MsgStream. A continuación, los nodos de datos consumen los datos, luego los convierten y almacenan los datos consumidos en el almacenamiento de objetos. La ruta de almacenamiento es un tipo de meta información que será registrada en etcd por los coordinadores de datos.

Diagrama de flujo

Dado que diferentes colecciones pueden compartir los mismos pchannels en el log broker, al consumir datos, los nodos de datos o los nodos de consulta necesitan juzgar a qué colección pertenecen los datos de un pchannel. Para resolver este problema, hemos introducido flowgraph en Milvus. Se encarga principalmente de filtrar los datos en un pchannel compartido por IDs de colección. Así, podemos decir que cada flowgraph gestiona el flujo de datos en un shard correspondiente (vchannel) en una colección.

Flowgraph en ruta de escritura

Creación de MsgStream

Cuando se escriben datos, el objeto MsgStream se crea en los dos escenarios siguientes:

• Cuando el proxy recibe una solicitud de inserción de datos, primero intenta obtener el mapeo entre vchannels y pchannels a través del coordinador raíz (root coord). A continuación, el proxy crea un objeto MsgStream.

Escenario 1

• Cuando se inicia el nodo de datos y lee la metainformación de los canales en etcd, se crea el objeto MsgStream.

Escenario 2

Lectura de datos

Ruta de lectura en Milvus

El flujo de trabajo general de la lectura de datos se ilustra en la imagen anterior. Las solicitudes de consulta se transmiten a través de DqRequestChannel a los nodos de consulta. Los nodos de consulta ejecutan las tareas de consulta en paralelo. Los resultados de consulta de los nodos de consulta pasan a través de gRPC y el proxy agrega los resultados y los devuelve al cliente.

Para ver más de cerca el proceso de lectura de datos, podemos ver que el proxy escribe peticiones de consulta en DqRequestChannel. A continuación, los nodos de consulta consumen los mensajes suscribiéndose a DqRequestChannel. Cada mensaje del DqRequestChannel se difunde para que todos los nodos de consulta suscritos puedan recibirlo.

Cuando los nodos de consulta reciben solicitudes de consulta, realizan una consulta local tanto de los datos por lotes almacenados en segmentos sellados como de los datos de flujo que se insertan dinámicamente en Milvus y se almacenan en segmentos crecientes. Después, los nodos de consulta tienen que agregar los resultados de la consulta tanto en segmentos sellados como en segmentos crecientes. Estos resultados agregados se transmiten al proxy a través de gRPC.

El proxy recoge todos los resultados de múltiples nodos de consulta y los agrega para obtener los resultados finales. A continuación, el proxy devuelve los resultados finales de la consulta al cliente. Dado que cada solicitud de consulta y sus correspondientes resultados están etiquetados con el mismo requestID único, el proxy puede averiguar qué resultados de consulta corresponden a qué solicitud de consulta.

Diagrama de flujo

Diagrama de flujo en la ruta de lectura

Al igual que en la ruta de escritura, también se introducen diagramas de flujo en la ruta de lectura. Milvus implementa la arquitectura Lambda unificada, que integra el procesamiento de los datos incrementales e históricos. Por lo tanto, los nodos de consulta también necesitan obtener datos de flujo en tiempo real. Del mismo modo, los flujos en la ruta de lectura filtran y diferencian los datos de diferentes colecciones.

Creación de MsgStream

Creación del objeto MsgStream en la ruta de lectura

Cuando se leen datos, el objeto MsgStream se crea en el siguiente escenario:

• En Milvus, los datos no pueden leerse a menos que estén cargados. Cuando el proxy recibe una solicitud de carga de datos, envía la solicitud al coordinador de consultas, que decide la forma de asignar los fragmentos a los distintos nodos de consulta. La información de asignación (es decir, los nombres de los vchannels y la correspondencia entre los vchannels y sus correspondientes pchannels) se envía a los nodos de consulta mediante una llamada a método o RPC (llamada a procedimiento remoto). Posteriormente, los nodos de consulta crean los objetos MsgStream correspondientes para consumir los datos.

Operaciones DDL

DDL significa lenguaje de definición de datos. Las operaciones DDL sobre metadatos pueden clasificarse en solicitudes de escritura y solicitudes de lectura. Sin embargo, estos dos tipos de solicitudes se tratan por igual durante el procesamiento de metadatos.

Las solicitudes de lectura de metadatos incluyen

• Consulta del esquema de recopilación
• Información de indexación de consultas Y más

Las solicitudes de escritura incluyen:

• Crear una colección
• Dar de baja una colección
• Crear un índice
• Eliminar un índice Y más

Las peticiones DDL se envían al proxy desde el cliente, y el proxy transmite estas peticiones en el orden recibido al coordenador raíz, que asigna una marca de tiempo a cada petición DDL y realiza comprobaciones dinámicas de las peticiones. El proxy gestiona cada solicitud de forma serial, es decir, una solicitud DDL cada vez. El proxy no procesará la siguiente solicitud hasta que termine de procesar la anterior y reciba los resultados del coord raíz.

Operaciones DDL.

Como se muestra en la ilustración anterior, hay K solicitudes DDL en la cola de tareas de Root coord. Las solicitudes DDL en la cola de tareas se ordenan según el orden en que las recibe el coordinador raíz. Así, ddl1 es la primera enviada al coordinador raíz y ddlK es la última de este lote. El coordinador raíz procesa las peticiones una a una en el orden temporal.

En un sistema distribuido, la comunicación entre los proxies y el coordinador raíz se realiza mediante gRPC. El root coord mantiene un registro del valor máximo de la marca de tiempo de las tareas ejecutadas para asegurar que todas las peticiones DDL se procesan en orden de tiempo.

Supongamos que hay dos proxies independientes, el proxy 1 y el proxy 2. Ambos envían peticiones DDL al servidor. Ambos envían peticiones DDL a la misma coordenada raíz. Sin embargo, uno de los problemas es que las peticiones anteriores no se envían necesariamente al coord raíz antes que las peticiones que recibe otro proxy más tarde. Por ejemplo, en la imagen anterior, cuando DDL_K-1 se envía al coordinador raíz desde el proxy 1, DDL_K desde el proxy 2 ya ha sido aceptada y ejecutada por el coordinador raíz. Según lo registrado por el coordinador raíz, el valor máximo de la marca de tiempo de las tareas ejecutadas en este punto es K. Por lo tanto, para no interrumpir el orden temporal, la solicitud DDL_K-1 será rechazada por la cola de tareas del coordinador raíz. Sin embargo, si el proxy 2 envía la solicitud DDL_K+5 al coordenador raíz en este punto, la solicitud será aceptada en la cola de tareas y se ejecutará más tarde de acuerdo con su valor de marca de tiempo.

Indexación

Creación de un índice

Al recibir solicitudes de creación de índices del cliente, el proxy realiza primero comprobaciones estáticas de las solicitudes y las envía al coordenador raíz. A continuación, el coordinador raíz persiste estas solicitudes de creación de índices en el metaalmacenamiento (etcd) y envía las solicitudes al coordinador de índices (coordinador de índices).

Creación de un índice.

Como se ilustra más arriba, cuando el coordinador del índice recibe solicitudes de creación de índices del coordinador raíz, primero persiste la tarea en etcd para el metaalmacenamiento. El estado inicial de la tarea de creación de índice es Unissued. El coordinador del índice mantiene un registro de la carga de tareas de cada nodo del índice y envía las tareas entrantes a un nodo del índice menos cargado. Al finalizar la tarea, el nodo de índice escribe el estado de la tarea, ya sea Finished o Failed en el metaalmacenamiento, que es etcd en Milvus. A continuación, el coordinador del índice sabrá si la tarea de creación del índice ha tenido éxito o ha fallado consultando el etcd. Si la tarea falla debido a que los recursos del sistema son limitados o al abandono del nodo de índice, el coordinador del índice volverá a iniciar todo el proceso y asignará la misma tarea a otro nodo de índice.

Abandono de un índice

Además, el coordinador de índices también se encarga de las solicitudes de eliminación de índices.

Baja de un índice.

Cuando el nodo raíz recibe una solicitud de eliminación de un índice por parte del cliente, primero marca el índice como "eliminado" y devuelve el resultado al cliente, notificándoselo al nodo índice. A continuación, el coordinador de índices filtra todas las tareas de indexación con IndexID y se eliminan las tareas que cumplen la condición.

La coroutina en segundo plano del coordinador de índices eliminará gradualmente todas las tareas de indexación marcadas como "descartadas" del almacenamiento de objetos (MinIO y S3). En este proceso interviene la interfaz recycleIndexFiles. Cuando se eliminan todos los archivos de índice correspondientes, la metainformación de las tareas de indexación eliminadas se elimina del metaalmacenamiento (etcd).

Acerca de la serie Deep Dive

Con el anuncio oficial de la disponibilidad general de Milvus 2.0, orquestamos esta serie de blogs Milvus Deep Dive para ofrecer una interpretación en profundidad de la arquitectura y el código fuente de Milvus. Los temas tratados en esta serie de blogs incluyen

• Visión general de la arquitectura de Milvus
• API y SDK de Python
• Procesamiento de datos
• Gestión de datos
• Consultas en tiempo real
• Motor de ejecución escalar
• Sistema de control de calidad
• Motor de ejecución vectorial