データのインポート（SDK）

このガイドでは、当社の SDK を使用し、bulk-writer API と bulk-import API でコレクションにデータをインポートする方法について説明します。

または、データの準備と Zilliz Cloud のコレクションへのデータのインポートの両方を扱う当社の高速なエンドツーエンドコースも参照できます。

Notes

Zilliz Cloud では現在、クラスターをホストするクラウドプロバイダーに関係なく、任意のオブジェクトストレージサービスから任意の Zilliz Cloud クラスターにデータをインポートできます。たとえば、AWS S3 バケットから GCP にデプロイされた Zilliz Cloud クラスターにデータをインポートできます。

低レイテンシで安定したエクスペリエンスを確保するには、ターゲットクラスターと同じプロバイダーおよび同じリージョンのバケットまたは blob コンテナーを使用することをお勧めします。

依存関係をインストールする

Python
Java

ターミナルで次のコマンドを実行し、pymilvus と minio をインストールするか、最新バージョンにアップグレードします。

shell

python3 -m pip install --upgrade pymilvus minio

Apache Maven の場合は、pom.xml の依存関係に以下を追加します：

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>

Gradle/Grails, の環境では、次のコマンドを実行します。

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'

準備済みデータを確認する

BulkWriter ツールを使用してデータを準備し、準備済みファイルのパスを取得したら、そのファイルを Zilliz Cloud コレクションにインポートする準備が整います。準備ができているかどうかを確認するには、次のようにします：

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 フィールド

StructArray フィールドは、ソースファイルのフォーマットごとに異なる形式で保持されます。インポートリーダーは、次のいずれかの形式を想定しています：

フォーマット	StructArray フィールドの形式
JSON、JSONL	フィールド値はオブジェクトのリストであり、Struct 要素ごとに 1 つのオブジェクトを持ち、サブフィールド名をキーとします：`"chunks": [{"text": "...", "section": "index", "emb": [0.1, 0.2]}, ...]`。
CSV	セルには、JSON フォーマットと同じ JSON テキストが格納されます。インポートジョブに設定された `nullkey` または JSON の `null` は、null である StructArray 値を示します。
Parquet	StructArray フィールドごとに、型が `list<struct<...>>` の列が 1 つあり、サブフィールドごとに 1 つの struct の子があります。スカラーのサブフィールドはスカラーの子、ベクトルのサブフィールドは `list<float32>` などの要素型のリストであり、配列のサブフィールドは `list<T>` の子です。
NumPy	サポートされていません。NumPy ファイルでは、トップレベルのフィールドごとに 1 ファイルが使用され、StructArray フィールドを表す方法はありません。StructArray フィールドを含むコレクションには、JSON、CSV、または Parquet ファイルを使用してください。

次のルールは、サポートされているすべてのフォーマットの StructArray 値に適用されます：

各 Struct 要素の内部では、chunks[text] のようなパスではなく、サブフィールド名を使用してください。chunks[text] という名前のフラットな列またはキーは拒否されます。
[null, {...}] のように、Struct 要素自体を null にすることはできません。StructArray 値全体が null になるのは、StructArray フィールドが null を許容する場合だけです。
コレクションのサブフィールドではないキーまたは Parquet の子（削除されたサブフィールドを含む）は無視されます。
要素が null を許容するサブフィールドでキーが見つからない場合、またはその値が null の場合、該当する Struct 要素の null 値としてインポートされるため、サブフィールドを追加する前に作成されたファイルも、そのサブフィールドなしでインポートできます。それ以外の場所に null がある場合はインポートに失敗します。配列のサブフィールドでは、null は null の内部配列を表し、内部配列内の null は、サブフィールドがそれを許容する場合、null の内部値を表します。
バックアップからの復元などの binlog インポートと、スナップショットソースからのインポートは、要素が null を許容するサブフィールドまたはスカラー配列のサブフィールドを持つコレクションでは拒否されます。

pymilvus BulkWriter は、要素が null を許容するサブフィールドと null を許容する StructArray 値を含め、サポートされているすべてのフォーマットで上記の StructArray 形式をサポートしています。ライターの add_row() 呼び出しでは、同じ行形式を使用してください。

スキーマ側の制限については、StructArray Limits を参照してください。

データをインポートする

データとコレクションの準備ができたら、オブジェクトストレージバケットやブロックストレージの blob コンテナーなどの外部ストレージを介して、データを特定のコレクションにインポートできます。

データをインポートする

外部ストレージを介してデータをインポートする場合は、次のようにします：

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

データのインポートを成功させるには、ターゲットコレクションの実行中または保留中のインポートジョブが 10,000 件未満であることを確認してください。

インポートの進行状況を確認する

指定した bulk-import ジョブの進行状況を確認できます。

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"
    }'

すべてのインポートジョブを一覧表示する

すべての bulk-import タスクについても確認する場合は、次のように list-import-jobs API を呼び出します：

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"
    }'

依存関係をインストールする​

準備済みデータを確認する​

ソースファイル内の StructArray フィールド​

データをインポートする​

データをインポートする​

インポートの進行状況を確認する​

すべてのインポートジョブを一覧表示する​

関連トピック​

依存関係をインストールする

準備済みデータを確認する

ソースファイル内の StructArray フィールド

データをインポートする

データをインポートする

インポートの進行状況を確認する

すべてのインポートジョブを一覧表示する

関連トピック