DataWorks support for Milvus data synchronization - DataWorks - Centro de documentación de Alibaba Cloud

DataWorks se conecta a Milvus como fuente de datos. Esto le permite leer y escribir en una base de datos vectorial de Milvus mediante tareas de sincronización por lotes.

Versiones compatibles de Milvus

Milvus 2.4.x
Milvus 2.5.x

Tipos de campo compatibles

A continuación se muestra una tabla que indica las correspondencias entre los tipos nativos de Milvus y las clasificaciones internas de tipos de DataWorks que utiliza Milvus Writer.

Tipo de dato de Milvus	Clasificación de tipo de DataWorks
Int8	LONG
Int16	LONG
Int32	LONG
Int64	LONG
Float	DOUBLE
Double	DOUBLE
FloatVector	DOUBLE
String	STRING
VarChar	STRING
SparseFloatVector	STRING
JSON	STRING
Array	STRING
Bool	BOOLEAN
BFloat16Vector	BYTES
Float16Vector	BYTES
BinaryVector	BYTES

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Una instancia de Milvus en ejecución y su URL de endpoint; por ejemplo, http://xxxx.milvus.aliyuncs.com:19530
Credenciales de autenticación de Milvus (nombre de usuario y contraseña)
Un espacio de trabajo de DataWorks con permiso para gestionar fuentes de datos y crear tareas de sincronización

Agregar una fuente de datos

Agregue una fuente de datos de Milvus a DataWorks antes de desarrollar una tarea de sincronización. Siga las instrucciones de Gestión de fuentes de datos. Las descripciones de los parámetros están disponibles directamente en la consola de DataWorks cuando agrega la fuente de datos.

Desarrollar una tarea de sincronización

DataWorks admite tareas de sincronización sin conexión que leen o escriben en colecciones de Milvus. El flujo de datos es el siguiente: el lector extrae los registros del origen → se aplica la correspondencia de tipos de campos → el escritor agrupa los registros por lotes en la colección de destino de Milvus.

Configurar una tarea de sincronización sin conexión para una sola tabla

Utilice cualquiera de los siguientes métodos:

Codeless UI configuration
Code editor configuration — consulte Appendix para ver un script de demostración completo y la referencia de parámetros

Apéndice: script de demostración y parámetros

Los siguientes scripts y tablas de parámetros aplican a las tareas de sincronización por lotes configuradas mediante el editor de código. Para conocer el formato general del script, consulte Configure a task in the code editor.

Lector

Script de demostración

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "milvusreader",
          "parameter": {
            "endpoint": "http://xxxx.milvus.aliyuncs.com:19530",
            "database": "default",
            "username": "root",
            "password": "xxxxxxx",
            "collection": "testColection",
            "column": [
              {
                "name": "id",
                "type": "Int64",
                "primaryKey": "true"
              },
              {
                "name": "int8col",
                "type": "Int8"
              },
              {
                "name": "int16col",
                "type": "Int16"
              }
            ]
          }
        },
        "writer": {
          "name": "Writer",
          "category": "writer",
          "stepType": "stream",
          "parameter": {}
        }
      }
    ],
    "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
        "throttle": false,
        "concurrent": 1,
        "channel": 1
      }
    }
  }
}

Parámetros del lector

Parámetro	Descripción	Obligatorio	Predeterminado
`collection`	La colección de la que se leerán los datos.	Sí	Ninguna
`column`	Los campos que se van a leer. Se configuran como un arreglo de objetos de campo, cada uno con `name` y `type`. Para el campo de clave principal, también establezca `primaryKey` en `"true"`. Para leer campos dinámicos, consulte las opciones de campos dinámicas más abajo.	Sí	Ninguna
`filter`	Condición de filtro equivalente a una cláusula WHERE. Para conocer la sintaxis, consulte la documentación de expresiones booleanas de Milvus.	No	Ninguna
`batchSize`	Número de registros que se leen por lote.	No	1024

Opciones de campos dinámicos para column

Para sincronizar todos los campos dinámicos como un solo campo JSON:

"column": [
  {
    "name": "dynamicName",
    "type": "json",
    "dynamicFileType": "allDynamicField"
  }
]

Para sincronizar un único campo dinámico, donde {singleDynamicName} es el nombre del campo en la colección:

"column": [
  {
    "name": "{singleDynamicName}",
    "type": "int",
    "dynamicFileType": "singleDynamicField"
  }
]

Escritor

Script de demostración

{
  "transform": false,
  "type": "job",
  "version": "2.0",
  "steps": [
    {
      "name": "Reader",
      "category": "reader",
      "stepType": "stream",
      "parameter": {}
    },
    {
      "name": "Writer",
      "category": "writer",
      "stepType": "milvus",
      "parameter": {
        "datasource": "zm_test",
        "collection": "test",
        "schemaCreateMode": "createIfNotExist",
        "enableDynamicSchema": true,
        "envType": 1,
        "column": [
          {
            "name": "floatv1",
            "type": "FloatVector",
            "dimension": "3"
          },
          {
            "name": "incol",
            "type": "Int16"
          }
        ],
        "writeMode": "insert",
        "batchSize": 1024
      }
    }
  ],
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "speed": {
      "concurrent": 2,
      "throttle": false
    }
  }
}

Parámetros del escritor

Parámetro	Descripción	Obligatorio	Predeterminado
`datasource`	Nombre de la fuente de datos configurada en DataWorks.	Sí	Ninguna
`collection`	Colección de destino en Milvus.	Sí	Ninguna
`schemaCreateMode`	Define cómo DataWorks gestiona la colección antes de la sincronización. Consulte Modos de creación de colecciones.	Sí	`createIfNotExist`
`column`	Campos de destino. Se configuran como un arreglo de objetos de campo. Cada campo requiere `name` y `type`. Los campos vectoriales también necesitan `dimension`; por ejemplo, `"dimension": "3"`.	Sí	Ninguna
`writeMode`	Comportamiento de escritura. Consulte Modos de escritura.	No	`upsert`
`partition`	Partición de destino. Deje este campo vacío para escribir en la partición `_default`.	No	`_default`
`batchSize`	Número de registros que se escriben por lote.	No	1024
`enableDynamicSchema`	Habilita el esquema dinámico cuando DataWorks crea la colección.	No	`true`

Modos de escritura

Modo	Comportamiento
`upsert`	En colecciones sin claves principales autoincrementales, actualiza una entidad según su clave principal. En colecciones con claves principales autoincrementales, reemplaza la clave principal por un valor generado automáticamente e inserta los datos.
`insert`	Inserta los datos sin comprobar si ya existen claves principales. Úselo en colecciones con claves principales autoincrementales, donde Milvus genera dichas claves de forma automática. Si utiliza `insert` en una colección sin claves principales autoincrementales, podrían generarse registros duplicados.

Modos de creación de colecciones

El parámetro schemaCreateMode controla qué hace DataWorks cuando la colección de destino no existe.

Modo	Comportamiento
`createIfNotExist`	Crea la colección a partir del parámetro `column` configurado y otros parámetros, y luego inicia la sincronización. Si la colección ya existe, la utiliza tal cual.
`Ignore`	Reporta un error si la colección no existe.
`recreate`	Elimina la colección existente y crea una nueva antes de cada ejecución de sincronización.