如何优化Spark与Milvus以高效处理大规模向量数据？

批量插入优化 #

• 调整批次大小：放弃20万条的大批次，Milvus单批次最优范围为1万-5万条。过大的批次会触发内存溢出，过小则会增加连接与IO开销。Spark侧需将DataFrame分区数调整为每个分区对应1万-5万条数据，例如执行df.repartition(50)（250万行对应50个分区，每个分区5万条）。
• 关闭自动刷新，手动批量提交：插入前关闭Milvus的自动刷新机制，减少频繁IO：

  data.set_properties({"auto_flush_interval": 0})
  

  所有批次插入完成后，手动触发刷新：

  data.flush()
  

• 避免资源竞争：Spark与Milvus不要抢占全部CPU资源，限制Spark的核心使用数，给Milvus预留足够的计算资源。

索引构建优化 #

• 选择适配高维向量的索引类型：20000维属于超高维向量，推荐使用IVF_PQ或IVF_SQ8索引（比IVF_FLAT更节省内存且速度更快）。若需精确查询，再考虑IVF_FLAT。
• 优化索引参数：针对250万条数据，设置nlist参数为1024-4096（平衡索引速度与查询精度），示例：

  data.create_index(
      field_name="vec",
      index_params={"metric_type": "L2", "index_type": "IVF_PQ", "params": {"nlist": 2048, "m": 16}}
  )
  

• 预处理数据：建索引前先执行data.flush()和data.compact()，规整数据存储结构，减少索引构建时的IO开销。
• 预留足够CPU资源：索引构建为CPU密集型任务，确保Milvus能分配到至少4核以上的CPU资源，避免与Spark进程抢占资源。

Spark配置调整 #

• 限制核心使用数：替换local[*]为具体核心数，例如local[12]，预留2-4核给Milvus与系统进程，避免CPU耗尽：

  spark = SparkSession.builder \
      .master("local[12]") \
      .appName("collab_filter_test_on_local") \
      .config("spark.driver.extraClassPath", '/data/notebook_files/clickhouse-native-jdbc-shaded-2.6.5.jar') \
      .config("spark.jars", "/data/notebook_files/spark-milvus-1.0.0-SNAPSHOT.jar") \
      .config("spark.driver.memory", "16g") \
      .config("spark.driver.maxResultSize", "8g") \
      .getOrCreate()
  

• 调整结果内存限制：设置spark.driver.maxResultSize为8G或更大，避免大分区数据处理时内存溢出。
• 配置连接器批次参数：使用Spark-Milvus写入时，明确指定批次大小：

  df.write.format("milvus") \
      .option("host", "localhost") \
      .option("port", "19530") \
      .option("collectionName", "data") \
      .option("batchSize", "40000") \
      .save()
  

Milvus配置调整（修改milvus.yaml） #

• 优化数据刷新与缓存：
  • dataCoord.flush.insertBufferSize: 64MB：调大插入缓冲区，减少flush次数
  • cache.cacheSize: 16GB：若机器总内存≥32G，设置为总内存的50%，加快数据加载速度
• 索引节点资源分配：
  • indexNode.resourceLimits.cpu: 8：给索引节点分配足够CPU核心，加速索引构建
• 集合配置：创建集合时指定分片数为1（单机部署无需多分片）：

  data = Collection('data', schema, shards_num=1)
  

• 异步加载集合：加载数据时使用异步模式，避免阻塞：

  data.load(replica_number=1, _async=True)
  

• 无需转换为numpy数组：Spark DataFrame中的vector列只要是Python列表、Scala Seq或Java List类型，Spark-Milvus连接器即可直接识别，无需额外转换为numpy数组。
• 严格校验向量维度：确保所有vector的长度严格等于20000，提前在Spark中过滤异常数据：

  from pyspark.sql.functions import udf
  from pyspark.sql.types import BooleanType
  
  def check_vec_length(vec):
      return len(vec) == 20000
  
  check_length_udf = udf(check_vec_length, BooleanType())
  valid_df = df.filter(check_length_udf(df.vector))
  

• 确保数据类型一致：vector中的元素必须是float类型，避免混入int或其他类型，否则插入会失败。

内容的提问来源于stack exchange，提问作者Fitzpatrick