文档中心

向量数据库VikingDB

数据(Data)

upsert_data

概述

upsert_data 用于在指定的数据集 Collection 内写入数据。指定写入的数据是一个数组，允许单次插入一条数据或者多条数据，单次最多可插入100条数据。
对 V1 版本的 upsert_data 接口进行了调整（仅影响含 vectorize/pipeline 配置的需要向量化数据集，已带 vector 类型字段的数据集不受影响），核心信息如下：

含 vectorize/pipeline 配置的数据集写入，embedding 计算将从实时同步转为后台异步执行，数据生效会有一定滞后；
upsert_data 接口不再返回 need_token_usage 参数信息。

请求参数

参数名	子参数	类型	是否必选	参数说明
Data Data 实例或者 Data 实例列表。	fields	map	是	指定写入的数据。单次写入的数据数目不超过100。每条数据作为一个 map，key 为字段名，value 为字段值。数据写入时 fields 长度最大为65535，超过限制时会返回报错 “fields data is too long, should be less than 65535”。不同字段类型的字段值格式如下： int64：格式是整型数值。 float：格式是浮点数值。 string：格式是字符串。 bool：格式是 true/false。 list：格式是字符串数组。 list：格式是整型数组。 vector：格式是向量（浮点数数组）。 sparse_vector：格式是 json 字典，k 为 string 类型，表示关键词的字面量，v 为 float 类型，表示该关键词的权重数值。 text：格式是 map<string, string>，当前支持 text 。 text：以 string 形式写入文本原始数据，如 {"text": "hello world"}。 image：当前支持写入tos路径。格式为`tos://{bucket}/{object_key}`。tos和VikingDB的region需要保持一致。
	ttl	int	否	数据过期时间，单位为秒。格式：0 和正整数。默认值：默认为0，表示数据不过期。当 ttl 设置为86400时，表示1天后数据自动删除。数据 ttl 删除，不会立刻更新到索引。
async_upsert		bool	否	对于“需要向量化”的数据集：本参数仅区分是否流式更新。 async_upsert=false，异步写入生效后触发索引流式更新 async_upsert=true，异步写入生效后仍不触发索引流式更新对于“已有向量数据”的数据集：异步写入开关。异步写入限流阈值为同步写入的 10 倍异步写入的数据不会实时同步写入 collection，滞后时间为分钟级别。可通过接口 `FetchDataInCollection` 来确认数据是否已经写入 collection 异步写入的数据不会触发索引的流式更新，索引同步时间为小时级别。可通过接口 `FetchDataInIndex`接口确认数据是否同步至 index

参数名

子参数

类型

是否必选

参数说明

Data
Data 实例或者 Data 实例列表。

fields

map

是

指定写入的数据。

单次写入的数据数目不超过100。
每条数据作为一个 map，key 为字段名，value 为字段值。
数据写入时 fields 长度最大为65535，超过限制时会返回报错 “fields data is too long, should be less than 65535”。
不同字段类型的字段值格式如下：
int64：格式是整型数值。
float：格式是浮点数值。
string：格式是字符串。
bool：格式是 true/false。
list：格式是字符串数组。
list：格式是整型数组。
vector：格式是向量（浮点数数组）。
sparse_vector：格式是 json 字典，k 为 string 类型，表示关键词的字面量，v 为 float 类型，表示该关键词的权重数值。
text：格式是 map<string, string>，当前支持 text 。
text：以 string 形式写入文本原始数据，如 {"text": "hello world"}。
image：当前支持写入tos路径。格式为tos://{bucket}/{object_key}。tos和VikingDB的region需要保持一致。

ttl

int

否

数据过期时间，单位为秒。

格式：0 和正整数。
默认值：默认为0，表示数据不过期。
当 ttl 设置为86400时，表示1天后数据自动删除。
数据 ttl 删除，不会立刻更新到索引。

async_upsert

bool

否

对于“需要向量化”的数据集：本参数仅区分是否流式更新。

async_upsert=false，异步写入生效后触发索引流式更新
async_upsert=true，异步写入生效后仍不触发索引流式更新

对于“已有向量数据”的数据集：异步写入开关。

异步写入限流阈值为同步写入的 10 倍
异步写入的数据不会实时同步写入 collection，滞后时间为分钟级别。可通过接口 FetchDataInCollection 来确认数据是否已经写入 collection
异步写入的数据不会触发索引的流式更新，索引同步时间为小时级别。可通过接口 FetchDataInIndex接口确认数据是否同步至 index

在执行upsert操作时选择上传数据的fields字段时，若选择“从向量化开始”，请根据需求上传text、image字段中的一个或一起上传，不要上传“vector”字段；若选择“已有向量数据”，请仅上传“vector”字段，不要上传“text”或 “image”字段。两者不能同时上传。
若需要上传image图多模态类型字段，请先将图片上传至TOS，并将图片的tos存储路径传入"image"字段。具体执行逻辑可参考如下流程图。详见【向量库】多模态搜索最佳实践

示例

运行示例前，请完成以下准备：
安装 VikingDB Python SDK，将完整代码保存为 example.py 后执行 python3 example.py。
将 <YOUR_HOST>、<YOUR_REGION> 和 <YOUR_SCHEME> 替换为地域和访问域名中与实例一致的服务地址、地域和协议。
将 <YOUR_AK>、<YOUR_SK> 替换为火山引擎控制台“密钥管理”中创建的 Access Key ID 和 Secret Access Key；请勿把真实凭证提交到代码仓库。
将 Collection、Index、Task 等资源占位符替换为控制台中对应资源的名称或接口返回的 ID。若示例包含 TOS 路径，请从 TOS 控制台确认桶名和对象路径，并使用 tos://<YOUR_BUCKET_NAME>/<YOUR_OBJECT_KEY> 或文档要求的路径格式。

请求参数

同步写入数据示例：

# 获取指定数据集，程序初始化时调用即可，无需重复调用
from volcengine.viking_db.CollectionClient import CollectionClient

collection_client = CollectionClient("<YOUR_COLLECTION_NAME>", "api-vikingdb.volces.com", "cn-beijing", "<YOUR_AK>", "<YOUR_SK>", "http")

此处为使用“已有向量数据”创建数据集，即已经对原始数据执行过embedding操作，因此上传“vector”类型数据。

# 构建向量
def gen_random_vector(dim):
    res = [0, ] * dim
    for i in range(dim):
        res[i] = random.random() - 0.5
    return res
   
field1 = {"doc_id": "11", "text_vector": gen_random_vector(12), "text_sparse_vector": {"hello": 0.34, "world": 0.03, "!": 0.11}, "like": 1, "price": 1.11,
          "author": ["gy"], "aim": True}
field2 = {"doc_id": "22", "text_vector": gen_random_vector(12), "text_sparse_vector": {"hi": 0.12, "there": 0.043, "!": 0.5},"like": 2, "price": 2.22,
          "author": ["gy", "xjq"], "aim": False}
data1 = Data(field1)
data2 = Data(field2)
datas = []
datas.append(data1)
datas.append(data2)
collection_client.upsert_data(datas)  


# 异步调用
async def upsert_data():
    def gen_random_vector(dim):
        res = [0, ] * dim
        for i in range(dim):
            res[i] = random.random() - 0.5
        return res


    collection_client = CollectionClient("<YOUR_COLLECTION_NAME>", "api-vikingdb.volces.com", "cn-beijing", "<YOUR_AK>", "<YOUR_SK>", "http")
    field1 = {"doc_id": "111", "text_vector": gen_random_vector(10), "like": 1, "price": 1.11,
             "author": ["gy"], "aim": True}
    field2 = {"doc_id": "222", "text_vector": gen_random_vector(10), "like": 2, "price": 2.22,
             "author": ["gy", "xjq"], "aim": False}
    field3 = {"doc_id": "333", "text_vector": gen_random_vector(10), "like": 3, "price": 3.33,
             "author": ["gy", "xjq"], "aim": False}
    field4 = {"doc_id": "444", "text_vector": gen_random_vector(10), "like": 4, "price": 4.44,
             "author": ["gy", "xjq"], "aim": False}
    data1 = Data(field1)
    data2 = Data(field2)
    data3 = Data(field3)
    data4 = Data(field4)
    datas = [data1, data2, data3, data4]
    await collection_client.async_upsert_data(datas)
asyncio.run(upsert_data())

此处为使用“从向量化开始”创建数据集，即原始数据未执行过embedding，因此直接上传原始数据“text”，”image“字段。

field1 = {"doc_id": "11", 
          "like": 1, 
          "text": "这是一段需要自动embedding的文本示例1",
          # 若需要上传多模态数据，请先将图片上传至TOS，并将图片存储路径传入"image"字段
          "image": "tos://your-bucket-name/path/to/image1.jpg",  # 图片存储路径
          "price": 1.11,
          "author": ["gy"],
           "aim": True}
field2 = {"doc_id": "22", 
          "like": 2, 
          "text": "这是一段需要自动embedding的文本示例2",
          # 若需要上传多模态数据，请先将图片上传至TOS，并将图片存储路径传入"image"字段
          "image": "tos://your-bucket-name/path/to/image2.jpg",  # 图片存储路径
          "price": 2.11,
          "author": ["zh"],
           "aim": True}
data1 = Data(field1)
data2 = Data(field2)
datas = []
datas.append(data1)
datas.append(data2)
collection_client.upsert_data(datas)

异步写入数据示例：

import multiprocessing
import struct, base64, uuid, tqdm, time
from volcengine.viking_db import *

queue = multiprocessing.Queue(maxsize=10)
event = multiprocessing.Event()
def consumer():
    """消费者函数：从队列中取出数据并处理"""
    vikingdb_service = VikingDBService()
    vikingdb_service.set_ak("ak")
    vikingdb_service.set_sk("sk")
    collection_client = CollectionClient("<YOUR_COLLECTION_NAME>", "api-vikingdb.volces.com", "cn-beijing", "<YOUR_AK>", "<YOUR_SK>", "http")
    items = []
    while not event.is_set() or not queue.empty():
        item = queue.get()
        items.append(item)
        if len(items) == 50:
            collection_client.upsert_data(items, async_upsert=True)
            items = []

    print("Consumer received event. Exiting...")

if __name__ == "__main__":
    # 创建消费者进程
    processors = []
    for i in range(10):
        p = multiprocessing.Process(target=consumer)
        p.start()
        processors.append(p)

    #  准备数据, 
    datas = []
    for i in range(100000):  
        # 压缩向量
        float_array = [0.124135132531424]*1024
        packed_data = struct.pack('f'*len(float_array), *float_array)
        s =  base64.b64encode(packed_data).decode()
        uuid4 = uuid.uuid4()  # 此处用户可修改为自己希望的id
        datas.append(Data({"id": str(uuid4), "text_vector": s}))
    
    for data in tqdm.tqdm(datas):
        queue.put(data)
    # 通知消费者停止工作
    event.set()
    for p in processors:
        p.join()

    print("Main process exiting...")

返回值

Python 调用执行上面的任务，执行成功无返回信息。

参数说明补充

async_upsert=true 表示异步写入：接口接收后后台处理，数据可见存在延迟；async_upsert=false 或省略表示同步写入：请求在本批写入处理完成后返回。已带向量字段的数据集使用同步模式；需要后台向量化时使用异步模式。

最近更新时间：2026.09.08 15:30:02

这个页面对您有帮助吗？

有用

无用

向量数据库VikingDB

请求参数 #

返回值 #

参数说明补充 #

请求参数

返回值

参数说明补充