Migrate data from a self-managed Qdrant cluster to an AnalyticDB for PostgreSQL instance

How it works

The first script reads data from Qdrant collections through the scroll API and writes the results to pipe-delimited CSV files. The second script bulk-loads those CSV files into AnalyticDB for PostgreSQL tables using the PostgreSQL COPY command.

Each Qdrant collection maps to a set of CSV files and a CREATE TABLE SQL statement during export. During import, each collection becomes a table in AnalyticDB for PostgreSQL.

Prerequisites

Before you begin, collect the following connection details:

Then confirm that the following are in place:

Step 1: Export data from the Qdrant cluster

1. Prepare the export.py script, the qdrant2csv.yaml configuration file, and an output directory.

   The export.py script contains the following content:

   import yaml
   import json
   from qdrant_client import QdrantClient
   import os
   from enum import IntEnum
   from tqdm import tqdm
   
   with open("./qdrant2csv.yaml", "r") as f:
       config = yaml.safe_load(f)
   
   print("configuration:")
   print(config)
   
   qdrant_config = config["qdrant"]
   
   class DataType(IntEnum):
       ID = 1
       FLOAT_VECTOR = 2
       JSON = 3
   
   def data_convert_to_str(data, dtype, delimeter):
       if dtype == DataType.ID:
           return str(data)
       elif dtype == DataType.FLOAT_VECTOR:
           return "{" + ", ".join(str(x) for x in data) + "}"
       elif dtype == DataType.JSON:
           return str(data).replace(delimeter, f"\\{delimeter}").replace("\"", "\\\"")
       Exception(f"Unsupported DataType {dtype}")
   
   def csv_write_rows(datum, fd, fields_types, delimiter="|"):
       for data in datum:
           for i in range(len(data)):
               data[i] = data_convert_to_str(data[i], fields_types[i], delimiter)
           fd.write(delimiter.join(data) + "\n")
   
   def csv_write_header(headers, fd, delimiter="|"):
       fd.write(delimiter.join(headers) + "\n")
   
   def dump_collection(collection_name: str):
       results = []
       file_cnt = 0
       print("connecting to qdrant...")
       client = QdrantClient(**qdrant_config)
   
       export_config = config["export"]
       tmp_path = os.path.join(export_config["output_path"], collection_name)
       if not os.path.exists(tmp_path):
           os.mkdir(tmp_path)
   
       # fetch info of collection
       fields_meta_list = ["id bigint"]
       fields_types = [DataType.ID]
       headers = ["id"]
       collection = client.get_collection(collection_name)
       total_num = collection.points_count
       if isinstance(collection.config.params.vectors, dict):
           # multi vectors
           for vec_name in collection.config.params.vectors.keys():
               fields_types.append(DataType.FLOAT_VECTOR)
               fields_meta_list.append(f"{vec_name} real[]")
               headers.append(vec_name)
       else:
           # single vector
           fields_types.append(DataType.FLOAT_VECTOR)
           fields_meta_list.append("vector real[]")
           headers.append("vector")
   
       fields_types.append(DataType.JSON)
       fields_meta_list.append("payload json")
       headers.append("payload")
   
       fields_meta_str = ','.join(fields_meta_list)
       create_table_sql = f"CREATE TABLE {collection_name} " \
                          f" ({fields_meta_str});"
   
       with open(os.path.join(export_config["output_path"], collection_name, "create_table.sql"), "w") as f_d:
           f_d.write(create_table_sql)
   
       print(create_table_sql)
   
       def write_to_csv_file(col_names, data):
           if len(results) == 0:
               return
           nonlocal file_cnt
           assert(file_cnt <= 1e9)
           output_file_name = os.path.join(export_config["output_path"], collection_name, f"{str(file_cnt).zfill(10)}.csv")
           with open(output_file_name, "w", newline="") as csv_file:
               # write header
               csv_write_header(col_names, csv_file)
               # write data
               csv_write_rows(data, csv_file, fields_types)
               file_cnt += 1
               results.clear()
   
       offset_id = None
   
       with tqdm(total=total_num, bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt}") as pbar:
           while True:
               res = client.scroll(collection_name=collection_name,
                                   limit=1000,
                                   offset=offset_id,
                                   with_payload=True,
                                   with_vectors=True)
   
               records = res[0]
               for record in records:
                   # append id
                   record_list = [record.id]
                   # append vectors
                   if isinstance(record.vector, dict):
                       # multi vector
                       for vector_name in headers[1:-1]:
                           record_list.append(record.vector[vector_name])
                   else:
                       # single vector
                       record_list.append(record.vector)
                   # append payload
                   record_list.append(json.dumps(record.payload, ensure_ascii=False))
                   results.append(record_list)
   
                   if len(results) >= export_config["max_line_in_file"]:
                       write_to_csv_file(headers, data=results)
   
                   pbar.update(1)
   
               if len(res) == 0 or len(res[0]) == 0 or res[1] is None:
                   # finished
                   break
               else:
                   offset_id = res[1]
   
       write_to_csv_file(headers, data=results)
   
   for name in config["export"]["collections"]:
       dump_collection(name)

   The qdrant2csv.yaml configuration file contains the following content:

   qdrant: # Connection settings for the Qdrant cluster.
       host: 'localhost'  # Host address of the Qdrant service.
       port: 6333        # REST port of the Qdrant service.
       grpc_port: 6434   # gRPC port of the Qdrant service.
       api_key: ''  # API key for Qdrant Cloud authentication. Leave blank if not required.
       url: ''      # Optional. Full URL including scheme, host, port, and prefix.
       location: '' # Optional. Set to "memory" to use in-memory mode. If a URL string, behaves the same as the url field. If blank, the host and port fields are used.
   
   export:
      collections:
       - 'test_collection'
       - 'multi'                 # List all Qdrant collections to export.
     max_line_in_file: 40000     # Maximum number of rows per output CSV file.
     output_path: './output'     # Directory where exported CSV files are saved.

   Note: The default Qdrant REST port is 6333 and the default gRPC port is 6334. Verify the port values for your deployment before running the export.

2. Place export.py, qdrant2csv.yaml, and the output directory in the same directory:

   ├── export.py
   ├── qdrant2csv.yaml
   └── output

3. Edit qdrant2csv.yaml to match your Qdrant cluster settings.

4. Run the export script:

   python export.py

   The script displays a progress bar during export. When export completes, the output directory contains one subdirectory per collection. Each subdirectory includes numbered CSV files and a create_table.sql file:

   .
   ├── export.py
   ├── qdrant2csv.yaml
   └── output
       ├── test_collection
       │   ├── 0000000000.csv
       │   ├── 0000000001.csv
       │   ├── 0000000002.csv
       │   └── create_table.sql
       └── multi
           ├── 0000000000.csv
           └── create_table.sql

Step 2: Import data into the AnalyticDB for PostgreSQL instance

1. Prepare the import.py script, the csv2adbpg.yaml configuration file, and the output directory from Step 1.

   The import.py script contains the following content:

   import psycopg2
   import yaml
   import glob
   import os
   
   if __name__ == "__main__":
       with open('csv2adbpg.yaml', 'r') as config_file:
           config = yaml.safe_load(config_file)
   
       print("current config:" + str(config))
   
       db_host = config['database']['host']
       db_port = config['database']['port']
       db_name = config['database']['name']
       schema_name = config['database']['schema']
       db_user = config['database']['user']
       db_password = config['database']['password']
       data_path = config['data_path']
   
       conn = psycopg2.connect(
           host=db_host,
           port=db_port,
           database=db_name,
           user=db_user,
           password=db_password,
           options=f'-c search_path={schema_name},public'
       )
   
       cur = conn.cursor()
   
       # check schema
       cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (schema_name,))
       existing_schema = cur.fetchone()
       if existing_schema:
           print(f"Schema {schema_name} already exists.")
       else:
           # create schema
           cur.execute(f"CREATE SCHEMA {schema_name}")
           print(f"Created schema: {schema_name}")
   
       for table_name in os.listdir(data_path):
           table_folder = os.path.join(data_path, table_name)
           print(f"Begin Process table: {table_name}")
           if os.path.isdir(table_folder):
               create_table_file = os.path.join(table_folder, 'create_table.sql')
               with open(create_table_file, 'r') as file:
                   create_table_sql = file.read()
               try:
                   cur.execute(create_table_sql)
               except psycopg2.errors.DuplicateTable as e:
                   print(e)
                   conn.rollback()
                   continue
               print(f"Created table: {table_name}")
   
               cnt = 0
               csv_files = glob.glob(os.path.join(table_folder, '*.csv'))
               for csv_file in csv_files:
                   with open(csv_file, 'r') as file:
                       copy_command = f"COPY {table_name} FROM STDIN DELIMITER '|' HEADER"
                       cur.copy_expert(copy_command, file)
                   cnt += 1
                   print(f"Imported data from: {csv_file} | {cnt}/{len(csv_files)} file(s) Done")
   
           conn.commit()
           print(f"Finished import table: {table_name}")
           print('#'*60)
   
       cur.close()
       conn.close()

   The csv2adbpg.yaml configuration file contains the following content:

   database:
       host: "192.16.XX.XX"         # Public endpoint of the AnalyticDB for PostgreSQL instance.
       port: 5432                   # Port of the AnalyticDB for PostgreSQL instance.
       name: "vector_database"      # Destination database name.
       user: "username"             # Database account username.
       password: ""                 # Password for the database account.
       schema: "public"             # Target schema. Created automatically if it does not exist.
   
   data_path: "./data"              # Path to the directory containing the exported CSV files.

2. Place import.py, csv2adbpg.yaml, and the data directory in the same directory:

   .
   ├── csv2adbpg.yaml
   ├── data
   │   ├── test_collection
   │   │   ├── 0000000000.csv
   │   │   ├── 0000000001.csv
   │   │   ├── 0000000002.csv
   │   │   └── create_table.sql
   │   └── multi
   │       ├── 0000000000.csv
   │       └── create_table.sql
   └── import.py

3. Edit csv2adbpg.yaml to match your AnalyticDB for PostgreSQL instance settings.

4. Run the import script:

   python import.py

   The script creates the target schema if it does not exist, creates tables from the generated SQL files, and bulk-loads data from the CSV files. If a table with the same name already exists, the script skips it and continues with the next collection.

Step 3: Verify the imported data

After the import completes, connect to the AnalyticDB for PostgreSQL instance and run the following checks.

1. Compare row counts. Run the following query for each imported table and compare the result against the points_count of the corresponding Qdrant collection:

   SELECT COUNT(*) FROM <table_name>;

2. Spot-check vectors and payloads:

   SELECT id, payload FROM <table_name> LIMIT 5;

   Verify that the id values are numeric (not UUID strings) and that the payload column contains valid JSON matching your source data.

3. If the row counts match and the sample data looks correct, the data import is complete.

Step 4: Rebuild indexes

The CSV-based migration does not preserve indexes from the source Qdrant cluster. After verifying the data, create vector indexes on the imported tables to enable efficient similarity search. For more information, see Create a vector index.

Limitations

References

• Qdrant documentation

• Create a vector index