from pymilvus import (    connections,    utility,    Collection,    DataType)import osimport jsonfrom pathlib import Path
fmt = "\n=== {:30} ===\n"
print(fmt.format("start connecting to Milvus"))host = os.environ.get('MILVUS_HOST', "localhost")print(fmt.format(f"Milvus host: {host}"))connections.connect("default", host=host, port="19530")
metrics_map = {  'COSINE': 'cosine',  'L2': 'euclidean',  'IP': 'dotproduct',}
dtype_map = {  DataType.BOOL: 'bool',  DataType.INT8: 'int',  DataType.INT16: 'int',  DataType.INT32: 'int',  DataType.INT64: 'int',
  DataType.FLOAT: 'float',  DataType.DOUBLE: 'float',
  DataType.STRING: 'str',  DataType.VARCHAR: 'str',}
def load_collection(collection_name: str) -> Collection:  has = utility.has_collection(collection_name)  print(f"Does collection hello_milvus exist in Milvus: {has}")  if not has:    return None
  collection = Collection(collection_name)        collection.load()    return collection  def export_collection_schema(collection, file: str):  schema = collection.schema.to_dict()  index = collection.indexes[0].to_dict()    export_schema = dict()    milvus_metric_type = index['index_param']['metric_type']  try:    export_schema['metrics'] = metrics_map[milvus_metric_type]  except:    raise Exception(f"milvus metrics_type{milvus_metric_type} not supported")  export_schema['fields_schema'] = {}    for field in schema['fields']:    if 'is_primary' in field and field['is_primary']:      continue        if field['name'] == index['field']:      # vector      if field['type'] == DataType.FLOAT_VECTOR:        export_schema['dtype'] = 'float'        export_schema['dimension'] = field['params']['dim']      else:        raise Exception(f"milvus dtype{field['type']} not supported yet")    else:      try:        # non-vector        export_schema['fields_schema'][field['name']] = dtype_map[field['type']]      except:        raise Exception(f"milvus dtype{field['type']} not supported yet")    with open(file, 'w') as file:    json.dump(export_schema, file, indent=4)    if __name__ == "__main__":  collection_name = "YOUR_MILVUS_COLLECTION_NAME"  collection = load_collection(collection_name)    dump_path_str = collection_name+'.dump'  dump_path = Path(dump_path_str)  dump_path.mkdir(parents=True, exist_ok=True)    schema_file = dump_path_str + "/schema.json"  export_collection_schema(collection, schema_file)

{    "metrics": "euclidean",    "fields_schema": {        "random": "float",        "var": "str"    },    "dtype": "float",    "dimension": 8}

from pymilvus import (    connections,    utility,    Collection,    DataType)import osimport jsonimport numpy as npfrom pathlib import Path
fmt = "\n=== {:30} ===\n"
print(fmt.format("start connecting to Milvus"))host = os.environ.get('MILVUS_HOST', "localhost")print(fmt.format(f"Milvus host: {host}"))connections.connect("default", host=host, port="19530")pk = "pk"vector_field_name = "vector"
def load_collection(collection_name: str) -> Collection:  has = utility.has_collection(collection_name)  print(f"Does collection hello_milvus exist in Milvus: {has}")  if not has:    return None
  collection = Collection(collection_name)        collection.load()    return collection  def export_partition_data(collection, partition_name, file: str):  batch_size = 10  output_fields=["pk", "random", "var", "embeddings"]  query_iter = collection.query_iterator(    batch_size=batch_size,    output_fields = output_fields,    partition_names=[partition_name]  )    export_file = open(file, 'w')    while True:    docs = query_iter.next()    if len(docs) == 0:      # close the iterator      query_iter.close()      break    for doc in docs:      new_doc = {}      new_doc_fields = {}      for k, v in doc.items():        if k == pk:          # primary key          new_doc['pk'] = str(v)        elif k == vector_field_name:          new_doc['vector'] = [float(k) for k in v]        else:          new_doc_fields[k] = v      new_doc['fields'] = new_doc_fields      json.dump(new_doc, export_file)      export_file.write('\n')        export_file.close() if __name__ == "__main__":  collection_name = "YOUR_MILVUS_COLLECTION_NAME"  collection = load_collection(collection_name)    pk = collection.schema.primary_field.name  vector_field_name = collection.indexes[0].field_name    dump_path_str = collection_name+'.dump'  dump_path = Path(dump_path_str)  dump_path.mkdir(parents=True, exist_ok=True)
  for partition in collection.partitions:    partition_name = partition.name    if partition_name == '_default':      export_path = dump_path_str + '/default.txt'    else:      export_path = dump_path_str + '/' + partition_name + ".txt"    export_partition_data(collection, partition_name, export_path)

from dashvector import Client, DashVectorException
from pydantic import BaseModelfrom typing import Dict, Typeimport json
dtype_convert = {  'int': int,  'float': float,  'bool': bool,  'str': str}
class Schema(BaseModel):  metrics: str  dtype: Type  dimension: int  fields_schema: Dict[str, Type]    @classmethod  def from_dict(cls, json_data):    metrics = json_data['metrics']    dtype = dtype_convert[json_data['dtype']]    dimension = json_data['dimension']    fields_schema = {k: dtype_convert[v] for k, v in json_data['fields_schema'].items()}    return cls(metrics=metrics, dtype=dtype, dimension=dimension, fields_schema=fields_schema)
def read_schema(schema_path) -> Schema:  with open(schema_path) as file:    json_data = json.loads(file.read())      return Schema.from_dict(json_data)
if __name__ == "__main__":  milvus_dump_path = f"{YOUR_MILVUS_COLLECTION_NAME}.dump"  milvus_dump_scheme_path = milvus_dump_path + "/schema.json"  schema = read_schema(milvus_dump_scheme_path)    client = dashvector.Client(    api_key='YOUR_API_KEY',    endpoint='YOUR_CLUSTER_ENDPOINT'  )    # create collection  rsp = client.create(name="YOUR_DASHVECTOR_COLLECTION_NAME",                       dimension=schema.dimension,                       metric=schema.metrics,                       dtype=schema.dtype,                      fields_schema=schema.fields_schema)  if not rsp:      raise DashVectorException(rsp.code, reason=rsp.message)

from dashvector import Client, DashVectorException, Doc
from pydantic import BaseModelfrom typing import Dict, Typeimport jsonimport globfrom pathlib import Path
def insert_data(collection, partition_name, partition_file):  if partition_name != 'default':    rsp = collection.create_partition(partition_name)    if not rsp:        raise DashVectorException(rsp.code, reason=rsp.message)      with open(partition_file) as f:    for line in f:      if line.strip():        json_data = json.loads(line)        rsp = collection.insert(          [            Doc(id=json_data['pk'], vector=json_data['vector'], fields=json_data['fields'])          ]        )        if not rsp:          raise DashVectorException(rsp.code, reason=rsp.message)  
if __name__ == "__main__":  milvus_dump_path = f"{YOUR_MILVUS_COLLECTION_NAME}.dump"
  client = dashvector.Client(    api_key='YOUR_API_KEY',    endpoint='YOUR_CLUSTER_ENDPOINT'  )    # create collection  collection = client.get("YOUR_DASHVECTOR_COLLECTION_NAME")    partition_files = glob.glob(milvus_dump_path+'/*.txt', recursive=False)  for partition_file in partition_files:    # create partition    partition_name = Path(partition_file).stem    insert_data(collection, partition_name, partition_file)