从MongoDB API版CosmosDB导海量数据到Spark DataFrame报40MB内存超限如何解决

Cosmos DB MongoDB API 对单条查询的服务端执行有固定40MB内存上限，对应错误码16501。该阈值无法在无实例管理权限的前提下通过调整配置、提升RUs修改，触发场景通常包括：

• 查询的过滤条件过宽，服务端扫描匹配的数据集规模超过内存承载上限
• Spark MongoDB Connector的分区策略配置错误，导致所有扫描请求聚合到单个查询窗口，压满服务端内存
• 查询管道中包含需要服务端临时缓存中间结果的操作（如全量排序、聚合）

• 修正查询管道的过滤逻辑，收窄单次扫描范围
  你当前代码中的查询管道存在硬编码参数错误：分区键值、id值均为固定字符串占位符，没有按实际传入的查询条件过滤。需将固定值匹配替换为范围匹配，按分片键的取值区间将大查询拆分为多次小范围扫描，每轮扫描的匹配结果规模控制在40MB阈值以内。
  可参考修正后的管道拼接逻辑：

  def getSubmissionsByIdRangePipeline(startId: String, endId: String, partitionKeyValue: String): Seq[Document] = {
    Seq(
      Document.parse(s"""{
        $match: {
          partition_key: "$partitionKeyValue",
          id: { $gte: "$startId", $lte: "$endId" }
        }
      }""")
    )
  }
  

• 调整Spark MongoDB Connector的分区配置，分布式拆分扫描请求
  你当前配置中将分片键设为固定常量MONGO_INPUT_SHARDKEY_VALUE，会导致连接器无法按字段值拆分扫描任务，所有数据都通过单个请求拉取，直接触发内存上限。需将分片键替换为实际的集合分片字段（通常是你过滤条件用到的id或partition_key），同时适当调大分区数，让每个分区对应的扫描请求规模更小。
  修正后的ReadConfig配置示例：

  def getReadConfig(uri: String, database: String, collection: String): ReadConfig = {
    ReadConfig(Map(
      "spark.mongodb.input.uri" -> uri,
      "spark.mongodb.input.database" -> database,
      "spark.mongodb.input.collection" -> collection,
      "readPreference.name" -> "secondaryPreferred",
      "spark.mongodb.input.partitioner" -> "MongoShardedPartitioner",
      "spark.mongodb.input.partitionerOptions.shardkey" -> "id",
      "spark.mongodb.input.partitionerOptions.numberOfPartitions" -> "20"
    ), Some(ReadConfig(sparkSession.sparkContext)))
  }
  

• 增加字段裁剪逻辑，降低单查询内存占用
  你当前仅需要映射Submission case class中定义的4个字段，原查询会拉取文档的全部字段，徒增服务端内存开销。可在查询管道中增加$project阶段，仅保留需要的字段，进一步降低单查询的内存占用：

  // 在原有的$match阶段后追加$project配置
  Document.parse("""{
    $project: {
      _id: 0,
      survey_id: 1,
      submitter_win: 1,
      answers: 1,
      submitted_on: 1
    }
  }""")
  

• 避免将需要服务端内存缓存的操作下推到Cosmos DB
  不要在查询管道中使用$sort、$group这类需要在服务端临时缓存全量中间结果的操作，这类操作会直接触发40MB内存阈值。如果需要排序、聚合类计算，将该类逻辑放到Spark侧的DataFrame转换中执行，不要推送到数据库端。
• 手动分批拉取拼接结果
  若上述调整后仍触发内存限制，可手动遍历分片键的取值区间，按20MB左右的单批结果规模逐批拉取数据，在Spark侧将多批RDD拼接为完整的DataFrame，替代连接器的自动全量拉取逻辑。

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