数据湖向量检索：Lance 和 Iceberg 上直接建索引，数据不搬

import json, os, time
from pymilvus import DataType, MilvusClient

client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")
extfs = {
    "cloud_provider": "aws",
    "region": "us-east-1",
    "access_key_id": os.environ["AWS_ACCESS_KEY_ID"],
    "access_key_value": os.environ["AWS_SECRET_ACCESS_KEY"],
}

def lake_collection(name, source, spec):
    schema = client.create_schema(external_source=source, external_spec=json.dumps(spec))
    # Map Milvus fields to existing source columns.
    schema.add_field(field_name="id", datatype=DataType.INT64, external_field="doc_id")
    schema.add_field(field_name="emb", datatype=DataType.FLOAT_VECTOR, dim=1024, external_field="embedding")
    schema.add_field(field_name="title", datatype=DataType.VARCHAR, max_length=1024, external_field="title")
    client.create_collection(collection_name=name, schema=schema)

# Lance: point at the dataset directory.
lake_collection("docs_lance", "s3://lake/docs.lance",
                {"format": "lance-table", "extfs": extfs})

# Iceberg: point at the table metadata file and pin a snapshot.
lake_collection("docs_iceberg", "s3://lake/docs/metadata/v1.metadata.json",
                {"format": "iceberg-table", "snapshot_id": 123456789, "extfs": extfs})

name = "docs_iceberg"
job_id = client.refresh_external_collection(collection_name=name)
deadline = time.monotonic() + 600
while (p := client.get_refresh_external_collection_progress(job_id=job_id)).state != "RefreshCompleted":
    if time.monotonic() > deadline:
        raise TimeoutError("External collection refresh timed out; inspect the job before retrying")
    if p.state == "RefreshFailed":
        raise RuntimeError(p.reason)
    time.sleep(1)

index_params = client.prepare_index_params()
index_params.add_index(field_name="emb", index_type="HNSW", metric_type="COSINE")
client.create_index(collection_name=name, index_params=index_params)
client.load_collection(collection_name=name)

hits = client.search(collection_name=name, data=[embed("rotate credentials")],  # your 1024-d embedding
                     anns_field="emb", limit=5, output_fields=["title"])
print(hits)