Import Data (SDK)

This guide helps you learn how to use our SDKs to import data into a collection with the bulk-writer and bulk-import APIs.

Alternatively, you can also refer to our fast-track end-to-end course which covers both data preparations and data import to Zilliz Cloud collections.

Notes

Zilliz Cloud now allows you to import data from any object storage service to any Zilliz Cloud cluster, regardless of the cloud provider hosting the clusters. For instance, you can import data from an AWS S3 bucket to a Zilliz Cloud cluster deployed on GCP.

You are advised to use a bucket or a blob container from the same provider and in the same region as the target cluster to ensure a low-latency, stable experience.

Install dependencies

Python
Java

Run the following command in your terminal to install pymilvus and minio or upgrade them to the latest version.

shell

python3 -m pip install --upgrade pymilvus minio

For Apache Maven, append the following to the pom.xml dependencies:

java
<dependency>
  <groupId>io.milvus</groupId>
  <artifactId>milvus-sdk-java</artifactId>
  <version>3.0.10</version>
</dependency>

<dependency>
  <groupId>io.milvus</groupId>
  <artifactId>milvus-sdk-java-bulkwriter</artifactId>
  <version>3.0.10</version>
</dependency>

<dependency>
    <groupId>io.minio</groupId>
    <artifactId>minio</artifactId>
    <version>8.5.9</version>
</dependency>

For Gradle/Grails, run the following

shell
implementation 'io.milvus:milvus-sdk-java:3.0.10'
implementation 'io.milvus:milvus-sdk-java-bulkwriter:3.0.10'
implementation 'io.minio:minio:8.5.9'

Check prepared data

Once you have prepared your data using the BulkWriter tool and got the path to the prepared files. You are ready to import them to a Zilliz Cloud collection. To check whether they are ready, do as follows:

Python
Java

python
from minio import Minio
from pymilvus import MilvusClient, DataType
from pymilvus.bulk_writer import bulk_import
import json
from pymilvus.bulk_writer import get_import_progress
from pymilvus.bulk_writer import list_import_jobs

# Third-party constants
ACCESS_KEY = "YOUR_ACCESS_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"
BUCKET_NAME = "YOUR_BUCKET_NAME"
REMOTE_PATH = "YOUR_REMOTE_PATH"

client = Minio(
    endpoint="storage.googleapis.com", # use 's3.amazonaws.com' for AWS S3
    access_key=ACCESS_KEY,
    secret_key=SECRET_KEY,
    secure=True
)

objects = client.list_objects(
    bucket_name=BUCKET_NAME,
    prefix=REMOTE_PATH,
    recursive=True
)

print([obj.object_name for obj in objects])

# Output
#
# [
#     "folder/1/claps.npy",
#     "folder/1/id.npy",
#     "folder/1/link.npy",
#     "folder/1/publication.npy",
#     "folder/1/reading_time.npy",
#     "folder/1/responses.npy",
#     "folder/1/title.npy",
#     "folder/1/vector.npy"
# ]

java
// Third-party constants
String ACCESS_KEY = "YOUR_ACCESS_KEY";
String SECRET_KEY = "YOUR_SECRET_KEY";
String BUCKET_NAME = "YOUR_BUCKET_NAME";
String REMOTE_PATH = "YOUR_REMOTE_PATH";

MinioClient minioClient = MinioClient.builder()
        .endpoint("storage.googleapis.com") // use 's3.amazonaws.com' for AWS S3
        .credentials(ACCESS_KEY, SECRET_KEY)
        .build();

Iterable<Result<Item>> results = minioClient.listObjects(
    ListObjectsArgs.builder().bucket(BUCKET_NAME).prefix(REMOTE_PATH).build()
);

for (Result<Item> result : results) {
    System.out.println(result.get().objectName());
}

// Output
// [[1.parquet]]

StructArray fields in source files

A StructArray field is carried differently by each source file format. The import reader expects one of the following shapes:

Format	StructArray field shape
JSON, JSONL	The field value is a list of objects, one per Struct element, keyed by subfield name: `"chunks": [{"text": "...", "section": "index", "emb": [0.1, 0.2]}, ...]`.
CSV	The cell holds the same JSON text as the JSON format. The `nullkey` configured for the import job or a JSON `null` marks a null StructArray value.
Parquet	One column per StructArray field with the type `list<struct<...>>`, one struct child per subfield. A scalar subfield is a scalar child, a vector subfield is a list of its element type, such as `list<float32>`, and an array subfield is a `list<T>` child.
NumPy	Not supported. NumPy files hold one file per top-level field and have no representation for a StructArray field. Use JSON, CSV, or Parquet files for collections that have a StructArray field.

The following rules apply to StructArray values in all supported formats:

Use subfield names inside each Struct element, not paths such as chunks[text]. Flat columns or keys named chunks[text] are rejected.
A Struct element itself cannot be null, such as [null, {...}]. A whole StructArray value can be null only when the StructArray field is nullable.
Keys or Parquet children that are not subfields of the collection, including dropped subfields, are ignored.
A missing key or a null value for an element-nullable subfield imports as a null value for that Struct element, so files written before a subfield was added import without it. A null anywhere else fails the import. For an array subfield, null is a null inner array, and null inside the inner array is a null inner value when the subfield allows it.
Binlog import, such as a backup restore, and snapshot-source import are rejected for collections that have element-nullable subfields or scalar array subfields.

pymilvus BulkWriter supports the StructArray shapes above in every supported format, including element-nullable subfields and nullable StructArray values. Use the same row shapes in the writer's add_row() calls.

For the schema-side limits, see StructArray Limits.

Import data

Once your data and collection are ready, you can import your data into a specific collection via an external storage, such as an object storage bucket and a block storage blob container.

Import data

If you prefer to import data via external storage, do as follows:

python
# Bulk-import your data from the prepared data files
CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com"
CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx"
API_KEY = ""
STORAGE_URL = ""
ACCESS_KEY = ""
SECRET_KEY = ""

res = bulk_import(
    api_key=API_KEY,
    url=CLOUD_API_ENDPOINT,
    cluster_id=CLUSTER_ID,
    collection_name="medium_articles",
    object_url=STORAGE_URL,
    access_key=ACCESS_KEY,
    secret_key=SECRET_KEY
)

print(res.json())

# Output
#
# {
#     "code": 0,
#     "data": {
#         "jobId": "9d0bc230-6b99-4739-a872-0b91cfe2515a"
#     }
# }

java
/**
 * The value of the URL is fixed.
 */
String CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
String CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
String API_KEY = "";
String STORAGE_URL = "";
String ACCESS_KEY = "";
String SECRET_KEY = "";
CloudImportRequest cloudImportRequest = CloudImportRequest.builder()
        .apiKey(API_KEY)
        .clusterId(CLUSTER_ID)
        .collectionName("medium_articles")
        .objectUrl(STORAGE_URL)
        .accessKey(ACCESS_KEY)
        .secretKey(SECRET_KEY)
        .build();
String bulkImportResult = BulkImportUtils.bulkImport(CLOUD_API_ENDPOINT, cloudImportRequest);
System.out.println(bulkImportResult);
JsonObject bulkImportObject = new Gson().fromJson(bulkImportResult, JsonObject.class);
String jobId = bulkImportObject.getAsJsonObject("data").get("jobId").getAsString();
System.out.println("Create a bulkImport task, job id: " + jobId);

// 0f7fe853-d93e-4681-99f2-4719c63585cc

go
ctx := context.Background()

opt := bulkwriter.NewCloudBulkImportOption(
    "https://api.cloud.zilliz.com",
    "medium_articles",
    "YOUR_API_KEY",
    "YOUR_OBJECT_URL",
    "inxx-xxxxxxxxxxxxxxx",
    "YOUR_ACCESS_KEY",
    "YOUR_SECRET_KEY",
)
resp, err := bulkwriter.BulkImport(ctx, opt)
if err != nil {
    log.Fatal(err)
}
fmt.Println(resp.Data.JobID)

rust
// The value of the URL is fixed.
let config = BulkImportConfig::new()
    .url("https://api.cloud.zilliz.com")
    .api_key("YOUR_API_KEY");
let bulk_import = BulkImport::new(&config)?;

let request = BulkImportRequest::builder()
    .collection_name("medium_articles")
    .cluster_id("inxx-xxxxxxxxxxxxxxx")
    .object_url("YOUR_OBJECT_URL")
    .access_key("YOUR_ACCESS_KEY")
    .secret_key("YOUR_SECRET_KEY")
    .build()?;

let resp = bulk_import.bulk_import(request).await?;
println!("{:?}", resp.job_id());

c++
// Note: milvus::BulkImport::CreateImportJobs() implements the import REST API,
// but as of v3.0.3 it does not accept the cloud object-storage parameters
// (objectUrl/accessKey/secretKey/clusterId) used in this section.

javascript

// Note: Cloud object-storage import is not supported in milvus-sdk-node as of v3.0.6.

csharp
// Bulk-import your data from the prepared data files
string CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
string CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
string API_KEY = "";
string STORAGE_URL = "";
string ACCESS_KEY = "";
string SECRET_KEY = "";

using var bulkImport = new BulkImport(new BulkImportConfig
{
    Url = CLOUD_API_ENDPOINT,
    ApiKey = API_KEY
});

var resp = await bulkImport.BulkImportAsync(new BulkImportRequest
{
    CollectionName = "medium_articles",
    ClusterId = CLUSTER_ID,
    ObjectUrl = STORAGE_URL,
    AccessKey = ACCESS_KEY,
    SecretKey = SECRET_KEY
});

Console.WriteLine(resp);
// {"code":0,"data":{"jobId":"9d0bc230-6b99-4739-a872-0b91cfe2515a"}}

bash
curl --request POST \
     --url "https://api.cloud.zilliz.com/v2/vectordb/jobs/import/create" \
     --header "Authorization: Bearer ${API_KEY}" \
     --header "Accept: application/json" \
     --header "Content-Type: application/json" \
     -d '{
        "clusterId": "inxx-xxxxxxxxxxxxxxx",
        "collectionName": "medium_articles",
        "objectUrl": "YOUR_OBJECT_URL",
        "accessKey": "YOUR_ACCESS_KEY",
        "secretKey": "YOUR_SECRET_KEY"
    }'

Notes

For successful data import, ensure the target collection has less than 10,000 running or pending import jobs.

Check import progress

You can check the progress of a specified bulk-import job.

python
## Zilliz Cloud constants
CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com"
CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx"
API_KEY = ""

# Get bulk-insert job progress
resp = get_import_progress(
    api_key=API_KEY,
    url=CLOUD_API_ENDPOINT,
    cluster_id=CLUSTER_ID,
    job_id="job-01fa0e5d42cjxudhpuehyp",
)

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

java
/**
 * The value of the URL is fixed.
 */
String CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
String CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
String API_KEY = "";
String jobId = "job-01fa0e5d42cjxudhpuehyp";
CloudDescribeImportRequest request = CloudDescribeImportRequest.builder()
    .apiKey(API_KEY)
    .clusterId(CLUSTER_ID)
    .jobId(jobId)
    .build();
String getImportProgressResult = BulkImportUtils.getImportProgress(CLOUD_API_ENDPOINT, request);
System.out.println("Get import progress, result: " + getImportProgressResult);

go
ctx := context.Background()

opt := bulkwriter.NewCloudGetImportProgressOption(
    "https://api.cloud.zilliz.com",
    "job-01fa0e5d42cjxudhpuehyp",
    "YOUR_API_KEY",
    "inxx-xxxxxxxxxxxxxxx",
)
resp, err := bulkwriter.GetImportProgress(ctx, opt)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("%+v\n", resp.Data)

rust
// The value of the URL is fixed.
let config = BulkImportConfig::new()
    .url("https://api.cloud.zilliz.com")
    .api_key("YOUR_API_KEY");
let bulk_import = BulkImport::new(&config)?;

let request = GetImportProgressRequest::builder()
    .job_id("job-01fa0e5d42cjxudhpuehyp")
    .cluster_id("inxx-xxxxxxxxxxxxxxx")
    .build()?;

let resp = bulk_import.get_import_progress(request).await?;
println!("{:?}", resp);

c++

// Note: milvus::BulkImport::GetImportJobProgress() implements the progress API,
// but as of v3.0.3 it does not send the cloud clusterId used in this section.

javascript

// Note: Cloud import progress is not supported in milvus-sdk-node as of v3.0.6.

csharp
// Zilliz Cloud constants
string CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
string CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
string API_KEY = "";

using var bulkImport = new BulkImport(new BulkImportConfig
{
    Url = CLOUD_API_ENDPOINT,
    ApiKey = API_KEY
});

// Get bulk-insert job progress
var resp = await bulkImport.GetImportProgressAsync(new GetImportProgressRequest
{
    JobId = "job-01fa0e5d42cjxudhpuehyp",
    ClusterId = CLUSTER_ID
});

Console.WriteLine(resp.Data);

bash
curl --request POST \
     --url "https://api.cloud.zilliz.com/v2/vectordb/jobs/import/get_progress" \
     --header "Authorization: Bearer ${API_KEY}" \
     --header "Accept: application/json" \
     --header "Content-Type: application/json" \
     -d '{
        "clusterId": "inxx-xxxxxxxxxxxxxxx",
        "jobId": "job-01fa0e5d42cjxudhpuehyp"
    }'

List all import jobs

If you also want to know about all bulk-import tasks, you can call the list-import-jobs API as follows:

python
## Zilliz Cloud constants
CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com"
CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx"
API_KEY = ""

# List bulk-insert jobs
resp = list_import_jobs(
    api_key=API_KEY,
    url=CLOUD_API_ENDPOINT,
    cluster_id=CLUSTER_ID
)

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

java
/**
 * The value of the URL is fixed.
 */
String CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
String CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
String API_KEY = "";
CloudListImportJobsRequest listImportJobsRequest = CloudListImportJobsRequest.builder()
        .apiKey(API_KEY)
        .clusterId(CLUSTER_ID).build();
String listImportJobsResult = BulkImportUtils.listImportJobs(CLOUD_API_ENDPOINT, listImportJobsRequest);
System.out.println(listImportJobsResult);

go
ctx := context.Background()

opt := bulkwriter.NewListImportJobsOption("https://api.cloud.zilliz.com", "")
opt.APIKey = "YOUR_API_KEY"
opt.ClusterID = "inxx-xxxxxxxxxxxxxxx"

resp, err := bulkwriter.ListImportJobs(ctx, opt)
if err != nil {
    log.Fatal(err)
}
fmt.Println(resp.Data.Records)

rust
// The value of the URL is fixed.
let config = BulkImportConfig::new()
    .url("https://api.cloud.zilliz.com")
    .api_key("YOUR_API_KEY");
let bulk_import = BulkImport::new(&config)?;

let request = ListImportJobsRequest::builder()
    .cluster_id("inxx-xxxxxxxxxxxxxxx")
    .build()?;

let resp = bulk_import.list_import_jobs(request).await?;
println!("{:?}", resp);

c++

// Note: milvus::BulkImport::ListImportJobs() implements the list-jobs API,
// but as of v3.0.3 it does not send the cloud clusterId used in this section.

javascript

// Note: Cloud import-jobs listing is not supported in milvus-sdk-node as of v3.0.6.

csharp
// Zilliz Cloud constants
string CLOUD_API_ENDPOINT = "https://api.cloud.zilliz.com";
string CLUSTER_ID = "inxx-xxxxxxxxxxxxxxx";
string API_KEY = "";

using var bulkImport = new BulkImport(new BulkImportConfig
{
    Url = CLOUD_API_ENDPOINT,
    ApiKey = API_KEY
});

// List bulk-insert jobs
var resp = await bulkImport.ListImportJobsAsync(new ListImportJobsRequest
{
    ClusterId = CLUSTER_ID
});

Console.WriteLine(resp.Data);

bash
curl --request POST \
     --url "https://api.cloud.zilliz.com/v2/vectordb/jobs/import/list" \
     --header "Authorization: Bearer ${API_KEY}" \
     --header "Accept: application/json" \
     --header "Content-Type: application/json" \
     -d '{
        "clusterId": "inxx-xxxxxxxxxxxxxxx"
    }'

Install dependencies​

Check prepared data​

StructArray fields in source files​

Import data​

Import data​

Check import progress​

List all import jobs​

Related topics​

Install dependencies

Check prepared data

StructArray fields in source files

Import data

Import data

Check import progress

List all import jobs

Related topics