DataWorks support for Milvus data synchronization - DataWorks

Versions de Milvus prises en charge

Milvus 2.4.x
Milvus 2.5.x

Types de champs pris en charge

Le tableau suivant présente les correspondances entre les types de données natifs de Milvus et les classifications de types internes de DataWorks utilisées par Milvus Writer.

Type de données Milvus	Classification du type 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érequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Une instance Milvus en cours d'exécution et son URL d'endpoint, par exemple http://xxxx.milvus.aliyuncs.com:19530
Les identifiants d'authentification Milvus (nom d'utilisateur et mot de passe)
Un espace de travail DataWorks disposant des autorisations nécessaires pour gérer les sources de données et créer des tâches de synchronisation

Ajouter une source de données

Ajoutez une source de données Milvus à DataWorks avant de développer une tâche de synchronisation. Suivez les instructions disponibles dans la rubrique Gestion des sources de données. Les descriptions des paramètres sont directement accessibles dans la console DataWorks lors de l'ajout de la source de données.

Développer une tâche de synchronisation

DataWorks prend en charge les tâches de synchronisation hors ligne qui lisent ou écrivent dans des collections Milvus. Le flux de données est le suivant : le lecteur extrait les enregistrements de la source → mappage des types de champs → le writer regroupe les enregistrements par lots dans la collection Milvus de destination.

Configurer une tâche de synchronisation hors ligne pour une table unique

Utilisez l'une des approches suivantes :

Configuration sans code via l'interface utilisateur
Configuration via l'éditeur de code — consultez l'Annexe pour une démonstration complète du script et la référence des paramètres

Annexe : Démonstration de script et paramètres

Les scripts et tableaux de paramètres suivants s'appliquent aux tâches de synchronisation par lots configurées via l'éditeur de code. Pour le format général du script, consultez la rubrique Configurer une tâche dans l'éditeur de code.

Reader

Démonstration 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
      }
    }
  }
}

Paramètres du reader

Paramètre	Description	Obligatoire	Valeur par défaut
`collection`	La collection à partir de laquelle lire les données.	Oui	Aucune
`column`	Les champs à lire. Configurez-les sous forme de tableau d'objets de champ, chacun contenant `name` et `type`. Pour le champ de clé primaire, définissez également `primaryKey` sur `"true"`. Pour lire les champs dynamiques, reportez-vous aux options de champ dynamique ci-dessous.	Oui	Aucune
`filter`	Une condition de filtre équivalente à une clause WHERE. Pour la syntaxe, consultez la documentation sur les expressions booléennes de Milvus.	Non	Aucune
`batchSize`	Le nombre d'enregistrements à lire par lot.	Non	1024

Options de champ dynamique pour column

Pour synchroniser tous les champs dynamiques en tant qu'un seul champ JSON :

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

Pour synchroniser un seul champ dynamique, où {singleDynamicName} représente le nom du champ dans la collection :

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

Writer

Démonstration 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
    }
  }
}

Paramètres du writer

Paramètre	Description	Obligatoire	Valeur par défaut
`datasource`	Le nom de la source de données tel que configuré dans DataWorks.	Oui	Aucune
`collection`	La collection de destination dans Milvus.	Oui	Aucune
`schemaCreateMode`	La manière dont DataWorks gère la collection avant la synchronisation. Consultez les Modes de création de collection.	Oui	`createIfNotExist`
`column`	Les champs de destination. Configurez-les sous forme de tableau d'objets de champ. Chaque champ nécessite `name` et `type`. Les champs vectoriels nécessitent également `dimension`, par exemple `"dimension": "3"`.	Oui	Aucune
`writeMode`	Le comportement d'écriture. Consultez les Modes d'écriture.	Non	`upsert`
`partition`	La partition de destination. Laissez vide pour écrire dans la partition `_default`.	Non	`_default`
`batchSize`	Le nombre d'enregistrements à écrire par lot.	Non	1024
`enableDynamicSchema`	Active le schéma dynamique lorsque DataWorks crée la collection.	Non	`true`

Modes d'écriture

Mode	Comportement
`upsert`	Pour les collections sans clés primaires auto-incrémentées : met à jour une entité en fonction de sa clé primaire. Pour les collections avec clés primaires auto-incrémentées : remplace la clé primaire par une valeur générée automatiquement et insère les données.
`insert`	Insère les données sans vérifier l'existence des clés primaires. À utiliser pour les collections avec clés primaires auto-incrémentées, où Milvus génère automatiquement les clés primaires. Si vous utilisez `insert` sur une collection sans clés primaires auto-incrémentées, des doublons peuvent apparaître.

Modes de création de collection

Le paramètre schemaCreateMode contrôle l'action entreprise par DataWorks lorsque la collection de destination n'existe pas.

Mode	Comportement
`createIfNotExist`	Crée la collection en fonction des paramètres `column` et autres configurés, puis démarre la synchronisation. Si la collection existe déjà, elle est utilisée telle quelle.
`Ignore`	Signale une erreur si la collection n'existe pas.
`recreate`	Supprime la collection existante et en crée une nouvelle avant chaque exécution de synchronisation.