跳到主要内容

使用 Iterator 导出数据

Zilliz Cloud 支持您将您的数据完整导出。本文将使用 Iterator 接口演示如何从 Zilliz Cloud Collection 中导出数据。

概述​

Milvus 的 Python 和 Java SDK 提供了一组 Iterator API，可按需迭代指定 Collection 中的数据。关于这些接口的具体情况，可查看使用 Iterators。

使用 Iterator 可获得如下收益：

  • 简化操作：免去复杂的 offset 和 limit 参数设置。

  • 提高效率：按需获取数据，享受更快速的数据检索体验。

  • 一致性：通过布尔过滤确保每次迭代都能维持数据集的一致大小。

您可使用 Iterator 接口全量或部分导出指定 Collection 中的数据。

📘说明

此功能支持与 Milvus v2.3.x 及之后版本兼容的 Zilliz Cloud 集群。

准备工作​

在如下步骤中，我们使用快速建表方法创建了一个 Collection, 并向其添加了 10,000 条随机数据。

创建 Collection​

from pymilvus import MilvusClient

CLUSTER_ENDPOINT = "YOUR_CLUSTER_ENDPOINT"
TOKEN = "YOUR_CLUSTER_TOKEN"

# 1. Set up a Milvus client
client = MilvusClient(
    uri=CLUSTER_ENDPOINT,
    token=TOKEN 
)

# 2. Create a collection
client.create_collection(
    collection_name="quick_setup",
    dimension=5,
)

插入随机数据​

# 3. Insert randomly generated vectors 
colors = ["green", "blue", "yellow", "red", "black", "white", "purple", "pink", "orange", "brown", "grey"]
data = []

for i in range(10000):
    current_color = random.choice(colors)
    current_tag = random.randint(1000, 9999)
    data.append({
        "id": i,
        "vector": [ random.uniform(-1, 1) for _ in range(5) ],
        "color": current_color,
        "tag": current_tag,
        "color_tag": f"{current_color}_{str(current_tag)}"
    })

print(data[0])

# Output
#
# {
#     "id": 0,
#     "vector": [
#         -0.5705990742218152,
#         0.39844925120642083,
#         -0.8791287928610869,
#         0.024163154953680932,
#         0.6837669917169638
#     ],
#     "color": "purple",
#     "tag": 7774,
#     "color_tag": "purple_7774"
# }

res = client.insert(
    collection_name="quick_setup",
    data=data,
)

print(res)

# Output
#
# {
#     "insert_count": 10000,
#     "ids": [
#         0,
#         1,
#         2,
#         3,
#         4,
#         5,
#         6,
#         7,
#         8,
#         9,
#         "(9990 more items hidden)"
#     ]
# }

使用 Iterator 导出数据​

如需使用 Iterator 导出数据，可按如下步骤操作：

  1. 在初始化 Iterator 对象时定义好检索参数（Search Parameters）及输出字段（Output Fields）。您还可以通过设置批大小（Batch Size）来控制每次迭代返回的数据量大小。

  2. 循环调用 Iterator 对象的 next() 方法来获取检索结果。

    • 如果该方法返回的是一个空数组，则循环中止。

    • 否则，您可以使用合适的方法将该方法返回的数据保存下来。比如，您可以将返回的数据依次添加到一个文件中，将其存放到某个数据库中，或者将其提供给下游应用消费。

  3. 在获取到所有满足条件的数据后，可调用 Iterator 对象的 close() 方法关闭 Iterator 对象。

如下代码演示了如何使用 QueryIterator 对象将导出的数据添加到一个文件中。

import json

# 6. Query with iterator

# Initiate an empty JSON file
with open('results.json', 'w') as fp:
    fp.write(json.dumps([]))

iterator = collection.query_iterator(
    batch_size=10,
    expr="color_tag like \"brown_8%\"",
    output_fields=["color_tag"]
)

while True:
    result = iterator.next()
    if not result:
        iterator.close()
        break
    
    # Read existing records and append the returns    
    with open('results.json', 'r') as fp:
        results = json.loads(fp.read())
        results += result
    
    # Save the result set    
    with open('results.json', 'w') as fp:
        fp.write(json.dumps(results))