跳到主要内容
版本：v2.6.x

导入数据

本页演示导入准备好的数据的步骤。

开始之前​

  • 您已经准备好数据并将其放入 Milvus 存储桶。

    如果没有，您应该先使用RemoteBulkWriter准备数据，并确保准备好的数据已经传输到与您的 Milvus 实例一起启动的 MinIO 实例上的 Milvus 数据桶中。有关详细信息，请参阅准备源数据。

  • 您已经使用用于准备数据的 Schema 创建了一个 Collection。如果没有，请参阅管理 Collection。

下面的代码片段使用给定的 Schema 创建了一个简单的 Collection。有关参数的更多信息，请参阅 create_schema()和 create_collection()SDK 参考资料。

以下代码片段使用给定的 Schema 创建一个简单 Collection。有关参数的更多信息，请参阅 createCollection()有关参数的更多信息，请参阅 SDK 参考资料中的

导入数据​

要导入准备好的数据，必须创建如下导入任务：

from pymilvus.bulk_writer import bulk_import

url = f"http://127.0.0.1:19530"

# Bulk-insert data from a set of JSON files already uploaded to the MinIO server
resp = bulk_import(
    url=url,
    collection_name="quick_setup",
    files=[['a1e18323-a658-4d1b-95a7-9907a4391bcf/1.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/2.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/3.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/4.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/5.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/6.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/7.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/8.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/9.parquet'],
           ['a1e18323-a658-4d1b-95a7-9907a4391bcf/10.parquet']],
)

job_id = resp.json()['data']['jobId']
print(job_id)

请求体包含两个字段：

  • collectionName

    目标 Collection 的名称。

  • files

    与 Milvus 实例一起启动的 MioIO 实例上相对于 Milvus 存储桶根路径的文件路径列表。可能的子列表如下：

    • JSON 文件

      如果准备的文件是 JSON 格式，则每个子列表都应包含单个准备的 JSON 文件的路径。

      [
          "/d1782fa1-6b65-4ff3-b05a-43a436342445/1.json"
      ],

    • Parquet 文件

      如果准备的文件是 Parquet 格式，则每个子列表都应包含单个准备的 parquet 文件的路径。

      [
          "/a6fb2d1c-7b1b-427c-a8a3-178944e3b66d/1.parquet"
      ]

可能的返回值如下：

{
    "code": 200,
    "data": {
        "jobId": "448707763884413158"
    }
}

检查导入进度​

获得导入任务 ID 后，可以按如下方式检查导入进度：

import json
from pymilvus.bulk_writer import get_import_progress

url = f"http://127.0.0.1:19530"

# Get bulk-insert job progress
resp = get_import_progress(
    url=url,
    job_id="453265736269038336",
)

print(json.dumps(resp.json(), indent=4))

可能的返回如下

{
    "code": 200,
    "data": {
        "collectionName": "quick_setup",
        "completeTime": "2024-05-18T02:57:13Z",
        "details": [
            {
                "completeTime": "2024-05-18T02:57:11Z",
                "fileName": "id:449839014328146740 paths:\"/8ca44f28-47f7-40ba-9604-98918afe26d1/1.parquet\" ",
                "fileSize": 31567874,
                "importedRows": 100000,
                "progress": 100,
                "state": "Completed",
                "totalRows": 100000
            },
            {
                "completeTime": "2024-05-18T02:57:11Z",
                "fileName": "id:449839014328146741 paths:\"/8ca44f28-47f7-40ba-9604-98918afe26d1/2.parquet\" ",
                "fileSize": 31517224,
                "importedRows": 100000,
                "progress": 100,
                "state": "Completed",
                "totalRows": 200000
            }
        ],
        "fileSize": 63085098,
        "importedRows": 200000,
        "jobId": "449839014328146739",
        "progress": 100,
        "state": "Completed",
        "totalRows": 200000
    }
}

列出导入任务​

您可以按如下方式列出相对于特定 Collection 的所有导入任务：

import json
from pymilvus.bulk_writer import list_import_jobs

url = f"http://127.0.0.1:19530"

# List bulk-insert jobs
resp = list_import_jobs(
    url=url,
    collection_name="quick_setup",
)

print(json.dumps(resp.json(), indent=4))

可能的值如下：

{
    "code": 200,
    "data": {
        "records": [
            {
                "collectionName": "quick_setup",
                "jobId": "448761313698322011",
                "progress": 50,
                "state": "Importing"
            }
        ]
    }
}

限制​

  • 每个导入文件的大小不得超过16 GB。

  • 每个导入请求的最大文件数不能超过1024。每个导入请求最多可导入 16GB 文件 * 1024 个文件 = 16TB 数据。

  • 并发导入请求的最大数量限制为1024。

  • 导入请求中只能指定一个 Partition 名称。如果没有指定 Partition 名称，数据将插入默认 Partition。此外，如果在目标 Collection 中设置了 Partition Key，则无法在导入请求中设置 Partition 名称。

限制条件​

导入数据前，请确保已确认以下 Milvus 行为方面的约束：

  • 有关加载行为的限制：

    • 如果在导入之前已经加载了一个 Collection，则可以在导入完成后使用refresh_load 函数加载新导入的数据。

  • 有关查询和搜索行为的限制：

    • 在导入任务状态为 "**已完成 "**之前，保证新导入的数据对查询和搜索是不可见的。

    • 一旦任务状态为完成、

      • 如果 Collection 尚未加载，可以使用load 函数加载新导入的数据。

      • 如果 Collection 已加载，则可调用load(is_refresh=True) 加载导入的数据。

  • 有关删除行为的限制：

    • 在导入任务状态为 "**已完成 "**之前，不保证删除成功。

    • 在任务状态为 "已完成"之后，删除将保证成功。

建议​

我们强烈建议使用多文件导入功能，该功能允许您在单个请求中上传多个文件。这种方法不仅简化了导入过程，还能显著提高导入性能。同时，通过合并上传，您可以减少用于数据管理的时间，提高工作流程的效率。