DataWorks的Milvus資料同步的能力支援情況 - DataWorks

Milvus資料來源為您提供寫入Milvus向量資料庫的通道，本文為您介紹DataWorks對Milvus資料同步的能力支援情況。

支援的Milvus版本

Milvus: 2.4.x
Milvus：2.5.x

支援的欄位類型

Milvus Writer針對Milvus的類型轉換列表，如下所示。

類型分類	Milvus資料類型
LONG	Int8、Int16、Int32、Int64
DOUBLE	Float、Double，FloatVector
STRING	String、VarChar、SparseFloatVector、JSON、Array
BOOLEAN	Bool
BYTES	BFloat16Vector、Float16Vector、BinaryVector

建立資料來源

在進行資料同步任務開發時，您需要在DataWorks上建立一個對應的資料來源，操作流程請參見建立並管理資料來源，詳細的配置參數解釋可在配置介面查看對應參數的文案提示。

資料同步任務開發

資料同步任務的配置入口和通用配置流程可參見下文的配置指導。

單表離線同步任務配置指導

操作流程請參見嚮導模式配置、指令碼模式配置。
指令碼模式配置的全量參數和指令碼Demo請參見下文的附錄：指令碼Demo與參數說明。

附錄：指令碼Demo與參數說明

離線任務指令碼配置方式

如果您配置離線任務時使用指令碼模式的方式進行配置，您需要按照統一的指令碼格式要求，在任務指令碼中編寫相應的參數，詳情請參見指令碼模式配置，以下為您介紹指令碼模式下資料來源的參數配置詳情。

Reader指令碼Demo

{
  "job": {
    "content": [
      {
        "reader": {
          "parameter": {
            "endpoint": "http://xxxx.milvus.aliyuncs.com:19530",
            "collection": "testColection",
            "database": "default",
            "password": "xxxxxxx",
            "username": "root",
            "column": [
              {
                "name": "id",
                "type": "Int64",
                "primaryKey": "true"
              },
              {
                "name": "int8col",
                "type": "Int8"
              },
              {
                "name": "int16col",
                "type": "Int16"
              }
            ]
          },
          "name": "milvusreader"
        },
        "writer": {
          "stepType": "stream",
          "parameter": {

          },
          "name": "Writer",
          "category": "writer"
        }
      }
    ],
    "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
        "throttle": false,
        "concurrent": 1,
        "channel": 1
      }
    }
  }
}

Reader指令碼參數

參數	描述	是否必選	預設值
collection	讀取Milvus的集合（表名）	是	無
batchSize	批量讀取的大小	否	1024
filter	資料讀取過濾條件，相當於where語句，配置方式參考：https://milvus.io/docs/zh/boolean.md	否	無
column	源端讀取的milvus欄位，支援兩種方式配置動態欄位同步：所有動態欄位作為一個JSON同步欄位。 `"cloumn":[{ "name":"dynamicName", "type":"json", "dynamicFileType":"allDynamicField" }]` 單動態欄位同步，{singleDynamicName}為集合中動態欄位名稱。 `"cloumn":[{ "name":"{singleDynamicName}", "type":"int", "dynamicFileType":"singleDynamicField" }]`	是	無

Writer指令碼Demo

{
  "transform": false,
  "type": "job",
  "version": "2.0",
  "steps": [
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType": "milvus",
      "parameter": {
        "schemaCreateMode": "createIfNotExist",     //集合建立模式
        "enableDynamicSchema": true,            //集合建立時是否開啟動態欄位
        "envType": 1,
        "datasource": "zm_test",
        "column": [  //同步欄位
          {
            "name": "floatv1",
            "type": "FloatVector",
            "dimension": "3"
          },
          {
            "name": "incol",
            "type": "Int16"
          }
        ],
        "writeMode": "insert",  //寫入方式
        "collection": "test",  //寫入集合
        "batchSize": 1024      // 單次寫入批量大小
      },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "speed": {
      "concurrent": 2,
      "throttle": false
    }
  }
}

Writer指令碼參數

指令碼參數名	描述	是否必選	預設值
datasource	資料來源名稱，指令碼模式支援添加資料來源，該配置項輸入的內容必須和添加的資料來源名稱保持一致。	是	無
collection	寫入Milvus的集合名。	是	無
partition	寫入Milvus集合的分區，不填寫入_default預設分區。	否	_default
column	Milvus 同步寫入Field欄位，配置為數組形式，單欄位資訊配置為JSON格式，內容包含： name： Field欄位名 type： Field欄位類型欄位屬性：如向量欄位的維度`"dimension":3`	是	無
writeMode	Milvus資料庫支援upsert和insert兩種寫入方式： upsert: 在非autoid表下根據主鍵更新 Collection 中的某個 Entity；autoid表Milvus 會將 Entity 中的主鍵替換為自動產生的主鍵，並插入資料。 insert: 多用於autoid表插入資料milvus自動產生主鍵，非autoid表下使用insert會導致資料重複。	否	upsert
batchSize	單次寫入Milvus的批量大小。	否	1024
schemaCreateMode	在同步前進行集合檢查，根據配置的模式進行集合操作，支援如下三種模式： createIfNotExist：檢查集合不存時，根據配置的column等資訊，建立對應集合進行同步。 Ignore：檢查集合不存時，報錯集合不存在 recreate：每一次同步，先刪除原集合，然後根據column等資訊重新建立集合進行同步。	是	createIfNotExist
enableDynamicSchema	建立集合是否啟用動態schema。	否	true