do_bulk_insert failed: key not found

[CharlesWu123]

do_bulk_insert failed info:

<Bulk insert state:
    - taskID          : 451021377873164808,
    - state           : Failed,
    - row_count       : 0,
    - infos           : {'failed_reason': 'read json file failed, path=14a76fbe-05bd-474a-9cb2-4e7317fbf3a5/1.json, err=key not found[key=14a76fbe-05bd-474a-9cb2-4e7317fbf3a5/1.json]: importing data failed', 'progress_percent': '0'},
    - id_ranges       : [],
    - create_ts       : 2024-07-09 18:00:19
>

The following is the partial code for processing and importing data

def get_bulk_writer(schema, bulk_save_dir='', remote=False, **kwargs):
    if isinstance(schema, dict):
        schema = CollectionSchema.construct_from_dict(schema)
    if remote:
        # 如果您的集群和 Milvus 2.4.2 及之后版本兼容
        # 需使用 `from pymilvus import LocalBulkWriter, BulkFileType`

        # 三方服务常量
        ACCESS_KEY = "minioadmin"
        SECRET_KEY = "minioadmin"
        BUCKET_NAME = "a-bucket"

        # 远程存储桶连接参数
        conn = RemoteBulkWriter.S3ConnectParam(
            endpoint="localhost:9000",
            # 腾讯云请使用 "cos.ap-beijing-1.myqcloud.com"
            access_key=ACCESS_KEY,
            secret_key=SECRET_KEY,
            bucket_name=BUCKET_NAME,
            secure=False
        )
        writer = RemoteBulkWriter(
            schema=schema,
            remote_path='/',
            connect_param=conn,
            file_type=BulkFileType.JSON
        )
    else:
        writer = LocalBulkWriter(
            schema=schema,
            local_path=bulk_save_dir,
            chunk_size=512 * 1024 * 1024,  # default value
            file_type=BulkFileType.JSON
        )
    return writer

def milvus_bulk_insert(collection_name, batch_files=None, using="default", worker_id=''):
    if not batch_files: batch_files = []
    logger.info(f'files: {batch_files}')
    connections.connect()
    # https://milvus.io/api-reference/pymilvus/v2.4.x/ORM/utility/do_bulk_insert.md
    task_ids = []
    for files in batch_files:
        task_id = utility.do_bulk_insert(
            collection_name=collection_name,
            files=files,
            using=using,
        )
        task_ids.append(task_id)
    # https://milvus.io/api-reference/pymilvus/v2.4.x/ORM/utility/get_bulk_insert_state.md
    while len(task_ids) > 0:
        time.sleep(1)
        for task_id in task_ids:
            state = utility.get_bulk_insert_state(task_id=task_id)
            if state.state == BulkInsertState.ImportFailed or state.state == BulkInsertState.ImportFailedAndCleaned:
                logger.error(f"The task {state.task_id} failed, reason: {state.failed_reason}")
                task_ids.remove(task_id)
            elif state.state == BulkInsertState.ImportCompleted:
                logger.error(f"The task {state.task_id} completed.")
                task_ids.remove(task_id)

bulk_writer = get_bulk_writer(schema, cur_bulk_save_dir, **kwargs)
for node in nodes:
    row_data = node_to_row(node)
    bulk_writer.append_row(row_data)
bulk_writer.commit()
milvus_bulk_insert(collection_name, using=using, batch_files=files, worker_id=worker_id)

The following is the printed log：

Is there a problem with which part of my processing

[yhmo]

Does this file exist in the minio?
0f9df******/1.json

[CharlesWu123]

I uploaded it through RemoteBulkWriter, and the file should be in minio. And I can also list the corresponding files through the following code.

    from minio import Minio

    # Third-party constants
    ACCESS_KEY = "minioadmin"
    SECRET_KEY = "minioadmin"
    BUCKET_NAME = "a-bucket"
    REMOTE_PATH = "/"

    client = Minio(
        endpoint="localhost:9000",
        # 腾讯云请使用 "cos.ap-beijing-1.myqcloud.com"
        access_key=ACCESS_KEY,
        secret_key=SECRET_KEY,
        secure=False
    )

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

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

[CharlesWu123]

But it is not a file, it is a dir. I don't know if this is normal.

[CharlesWu123]

Solved. This is my problem. Previously, I accidentally changed the value of common.storageType in milvus.yml to local, which caused this problem. Just restore this value to remote.