데이터 가져오기
이 페이지에서는 준비된 데이터를 가져오는 절차를 설명합니다.
시작하기 전에
이미 데이터를 준비하여 Milvus 버킷에 넣었습니다.
그렇지 않은 경우, 먼저 RemoteBulkWriter를 사용하여 데이터를 준비하고, 준비된 데이터가 Milvus 인스턴스와 함께 시작된 MinIO 인스턴스의 Milvus 버킷으로 이미 전송되었는지 확인해야 합니다. 자세한 내용은 소스 데이터 준비하기를 참조하세요.
데이터 준비에 사용하는 스키마로 컬렉션을 이미 만들었습니다. 그렇지 않은 경우 컬렉션 관리를 참조하세요.
다음 코드 조각은 주어진 스키마로 간단한 컬렉션을 만듭니다. 매개 변수에 대한 자세한 내용은 create_schema() 및 create_collection() 를 참조하세요.
다음 코드 스니펫은 주어진 스키마로 간단한 컬렉션을 만듭니다. 매개 변수에 대한 자세한 내용은 SDK 참조에서 createCollection() 를 참조하세요.
데이터 가져오기
준비된 데이터를 가져오려면 다음과 같이 가져오기 작업을 생성해야 합니다:
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)
private static String bulkImport(List<List<String>> batchFiles) throws InterruptedException {
MilvusImportRequest milvusImportRequest = MilvusImportRequest.builder()
.collectionName("quick_setup")
.files(batchFiles)
.build();
String bulkImportResult = BulkImport.bulkImport("http://localhost:19530", milvusImportRequest);
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 bulkInert task, job id: " + jobId);
return jobId;
}
public static void main(String[] args) throws Exception {
List<List<String>> batchFiles = uploadData();
String jobId = bulkImport(batchFiles);
}
export MILVUS_URI="localhost:19530"
curl --request POST "http://${MILVUS_URI}/v2/vectordb/jobs/import/create" \
--header "Content-Type: application/json" \
--data-raw '{
"files": [
[
"/8ca44f28-47f7-40ba-9604-98918afe26d1/1.parquet"
],
[
"/8ca44f28-47f7-40ba-9604-98918afe26d1/2.parquet"
]
],
"collectionName": "quick_setup"
}'
요청 본문에는 두 개의 필드가 포함됩니다:
collectionName대상 컬렉션의 이름.
filesMilvus 인스턴스와 함께 시작된 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))
private static void getImportProgress(String jobId) {
while (true) {
System.out.println("Wait 5 second to check bulkInsert job state...");
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
break;
}
MilvusDescribeImportRequest request = MilvusDescribeImportRequest.builder()
.jobId(jobId)
.build();
String getImportProgressResult = BulkImport.getImportProgress("http://localhost:19530", request);
JsonObject getImportProgressObject = new Gson().fromJson(getImportProgressResult, JsonObject.class);
String state = getImportProgressObject.getAsJsonObject("data").get("state").getAsString();
String progress = getImportProgressObject.getAsJsonObject("data").get("progress").getAsString();
if ("Failed".equals(state)) {
String reason = getImportProgressObject.getAsJsonObject("data").get("reason").getAsString();
System.out.printf("The job %s failed, reason: %s%n", jobId, reason);
break;
} else if ("Completed".equals(state)) {
System.out.printf("The job %s completed%n", jobId);
break;
} else {
System.out.printf("The job %s is running, state:%s progress:%s%n", jobId, state, progress);
}
}
}
public static void main(String[] args) throws Exception {
List<List<String>> batchFiles = uploadData();
String jobId = bulkImport(batchFiles);
getImportProgress(jobId);
}
export MILVUS_URI="localhost:19530"
curl --request POST "http://${MILVUS_URI}/v2/vectordb/jobs/import/describe" \
--header "Content-Type: application/json" \
--data-raw '{
"jobId": "449839014328146739"
}'
가능한 응답은 다음과 같습니다:
{
"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
}
}
가져오기 작업 목록
다음과 같이 특정 컬렉션과 관련된 모든 가져오기 작업을 나열할 수 있습니다:
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))
private static void listImportJobs() {
MilvusListImportJobsRequest listImportJobsRequest = MilvusListImportJobsRequest.builder().collectionName("quick_setup").build();
String listImportJobsResult = BulkImport.listImportJobs("http://localhost:19530", listImportJobsRequest);
System.out.println(listImportJobsResult);
}
public static void main(String[] args) throws Exception {
listImportJobs();
}
export MILVUS_URI="localhost:19530"
curl --request POST "http://${MILVUS_URI}/v2/vectordb/jobs/import/list" \
--header "Content-Type: application/json" \
--data-raw '{
"collectionName": "quick_setup"
}'
가능한 값은 다음과 같습니다:
{
"code": 200,
"data": {
"records": [
{
"collectionName": "quick_setup",
"jobId": "448761313698322011",
"progress": 50,
"state": "Importing"
}
]
}
}
제한 사항
각 가져오기 파일 크기는 16GB를 초과할 수 없습니다.
가져오기 요청의 최대 개수는 1024개로 제한됩니다.
가져오기 요청당 최대 파일 수는 1024개를 초과할 수 없습니다.
가져오기 요청에는 하나의 파티션 이름만 지정할 수 있습니다. 파티션 이름을 지정하지 않으면 데이터가 기본 파티션에 삽입됩니다. 또한 대상 컬렉션에 파티션 키를 설정한 경우에는 가져오기 요청에서 파티션 이름을 설정할 수 없습니다.
제약 조건
데이터를 가져오기 전에 다음 Milvus 동작에 대한 제약 조건을 확인했는지 확인하세요:
로드 동작에 관한 제약 조건:
- 가져오기 전에 컬렉션이 이미 로드된 경우, 가져오기가 완료된 후
refresh_load함수를 사용하여 새로 가져온 데이터를 로드할 수 있습니다.
- 가져오기 전에 컬렉션이 이미 로드된 경우, 가져오기가 완료된 후
쿼리 및 검색 동작 관련 제약 조건:
가져오기 작업 상태가 완료되기 전에는 새로 가져온 데이터가 쿼리 및 검색에 보이지 않도록 보장됩니다.
작업 상태가 완료되면,
컬렉션이 로드되지 않은 경우
load함수를 사용하여 새로 가져온 데이터를 로드할 수 있습니다.컬렉션이 이미 로드된 경우
load(is_refresh=True)을 호출하여 가져온 데이터를 로드할 수 있습니다.
삭제 동작에 관한 제약 조건
가져오기 작업 상태가 완료되기 전에는 삭제가 보장되지 않으며 삭제가 성공할 수도 있고 실패할 수도 있습니다.
작업 상태가 완료된 후의 삭제는 성공이 보장됩니다.
권장 사항
한 번의 요청으로 여러 개의 파일을 업로드할 수 있는 다중 파일 가져오기 기능을 사용하는 것을 적극 권장합니다. 이 방법은 가져오기 프로세스를 간소화할 뿐만 아니라 가져오기 성능도 크게 향상시킵니다. 한편, 업로드를 통합하면 데이터 관리에 소요되는 시간을 줄이고 워크플로우를 더욱 효율적으로 만들 수 있습니다.
