DataWorks support for Milvus data synchronization - DataWorks: Central de Documentação do Alibaba Cloud

Versões do Milvus compatíveis

Milvus 2.4.x
Milvus 2.5.x

Tipos de campo compatíveis

A tabela a seguir lista os mapeamentos entre os tipos nativos do Milvus e as classificações de tipos internos do DataWorks usadas pelo Milvus Writer.

Tipo de dados do Milvus	Classificação de tipo do 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

Pré-requisitos

Antes de começar, verifique se você tem:

Uma instância do Milvus em execução e o URL do endpoint correspondente, por exemplo http://xxxx.milvus.aliyuncs.com:19530
Credenciais de autenticação do Milvus (nome de usuário e senha)
Um workspace do DataWorks com permissão para gerencie fontes de dados e crie tarefas de sincronização

Adicionar uma fonte de dados

Adicione uma fonte de dados do Milvus ao DataWorks antes de desenvolver uma tarefa de sincronização. Siga as instruções em Gerenciamento de fontes de dados. As descrições dos parâmetros estão disponíveis diretamente no console do DataWorks ao adicionar a fonte de dados.

Desenvolver uma tarefa de sincronização

O DataWorks oferece suporte a tarefas de sincronização offline que leem ou gravam em coleções do Milvus. O fluxo de dados é: o reader extrai registros da origem → mapeamento de tipos de campos → o writer agrupa os registros na coleção de destino do Milvus.

Configure uma tarefa de sincronização offline para uma única tabela

Use uma das seguintes abordagens:

Configuração pela interface visual sem código
Configuração pelo editor de código — consulte o Apêndice para obter uma demonstração completa de script e referência de parâmetros

Apêndice: Demonstração de script e parâmetros

Os scripts e as tabelas de parâmetros a seguir aplicam-se a tarefas de sincronização em lote configuradas pelo editor de código. Para o formato geral do script, consulte Configure uma tarefa no editor de código.

Reader

Demonstração de script

{
  "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 do Reader

Parâmetro	Descrição	Obrigatório	Padrão
`collection`	Coleção de origem da leitura.	Sim	Nenhum
`column`	Campos a serem lidos. Configure como um array de objetos de campo, cada um com `name` e `type`. Para o campo de chave primária, defina também `primaryKey` como `"true"`. Para ler campos dinâmicos, consulte as opções de campo dinâmico abaixo.	Sim	Nenhum
`filter`	Condição de filtro equivalente a uma cláusula WHERE. Para sintaxe, consulte a documentação de expressões booleanas do Milvus.	Não	Nenhum
`batchSize`	Número de registros a serem lidos por lote.	Não	1024

Opções de campo dinâmico para column

Para sincronizar todos os campos dinâmicos como um único campo JSON:

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

Para sincronizar um único campo dinâmico, onde {singleDynamicName} é o nome do campo na coleção:

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

Writer

Demonstração de script

{
  "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 do Writer

Parâmetro	Descrição	Obrigatório	Padrão
`datasource`	Nome da fonte de dados conforme configurado no DataWorks.	Sim	Nenhum
`collection`	Coleção de destino no Milvus.	Sim	Nenhum
`schemaCreateMode`	Define como o DataWorks trata a coleção antes da sincronização. Consulte Modos de criação de coleção.	Sim	`createIfNotExist`
`column`	Campos de destino. Configure como um array de objetos de campo. Cada campo requer `name` e `type`. Campos vetoriais também exigem `dimension`, por exemplo `"dimension": "3"`.	Sim	Nenhum
`writeMode`	Comportamento de gravação. Consulte Modos de gravação.	Não	`upsert`
`partition`	Partição de destino. Deixe em branco para gravar na partição `_default`.	Não	`_default`
`batchSize`	Número de registros a serem gravados por lote.	Não	1024
`enableDynamicSchema`	Habilita o schema dinâmico quando o DataWorks cria a coleção.	Não	`true`

Modos de gravação

Modo	Comportamento
`upsert`	Para coleções sem chaves primárias de incremento automático: atualiza uma entidade com base em sua chave primária. Para coleções com chaves primárias de incremento automático: substitui a chave primária por um valor gerado automaticamente e insere os dados.
`insert`	Insere dados sem verificar a existência de chaves primárias. Use para coleções com chaves primárias de incremento automático, nas quais o Milvus gera as chaves primárias automaticamente. Se você usar `insert` em uma coleção sem chaves primárias de incremento automático, podem ocorrer registros duplicados.

Modos de criação de coleção

O parâmetro schemaCreateMode controla a ação do DataWorks quando a coleção de destino não existe.

Modo	Comportamento
`createIfNotExist`	Cria a coleção com base nos parâmetros de `column` e outros configurados e inicia a sincronização. Caso a coleção já exista, usa-a sem alterações.
`Ignore`	Retorna um erro se a coleção não existir.
`recreate`	Exclui a coleção existente e cria uma nova antes de cada execução de sincronização.