从嵌套MongoDB构建DataFrame：如何添加runTime字段

首先先修正下你提供的MongoDB文档的语法小问题（避免歧义）：

{ 
  "_id" : ObjectId("1"), 
  "createdDateTime" : ISODate("2016-11-23T12:21:10.726+0000"), 
  "data" : { 
    "price" : [ 
      { "timeStamp" : ISODate("2016-11-23T23:00:00.000+0000"), "value" : 32.92 }, 
      { "timeStamp" : ISODate("2016-11-24T00:00:00.000+0000"), "value" : 31.54 }, 
      { "timeStamp" : ISODate("2016-11-24T01:00:00.000+0000"), "value" : 29.14 }
    ]
  },
  "Type": "f",
  "runTime" : ISODate("2016-11-24T15:26:00.000+0000") 
}

你的原代码只提取了data.price数组转成DataFrame，但没把runTime带进来。核心思路是：先获取包含runTime和price的文档片段，然后给price数组的每一条数据都添加上对应的runTime值，再转成DataFrame。

下面是修改后的代码：

query_fc = {'Type': 'f', 'runTime': {"$lte": model_dt_till}}
# 只投影需要的字段：runTime和data.price，提升查询效率
projection = {'runTime': 1, 'data.price': 1, '_id': 0}
# 获取目标文档，按createdDateTime倒序取最新的
target_doc = self._collection.find_one(query_fc, projection=projection, sort=[('createdDateTime', -1)])

if target_doc:
    # 提取runTime值
    run_time = target_doc['runTime']
    # 获取price数组，给每个元素添加runTime字段
    price_data = target_doc['data']['price']
    for item in price_data:
        item['runTime'] = run_time
    # 转成DataFrame
    df_f = pd.DataFrame(price_data)
else:
    # 处理没有匹配文档的情况
    df_f = pd.DataFrame(columns=['timeStamp', 'value', 'runTime'])

代码解释： #

• 投影优化：通过projection参数指定只返回runTime、data.price，排除_id和其他无关字段，减少数据传输量，提升查询速度。
• 关联runTime：遍历price数组的每一条记录，把当前文档的runTime添加为每条记录的新字段，这样每条价格数据都能对应上它所属的runTime。
• 健壮性处理：增加了target_doc是否存在的判断，避免查询不到文档时出现KeyError。

如果你的场景中需要处理多个匹配文档（而不是只取最新的一个），可以把find_one换成find，然后循环处理每个文档，最后合并所有DataFrame：

query_fc = {'Type': 'f', 'runTime': {"$lte": model_dt_till}}
projection = {'runTime': 1, 'data.price': 1, '_id': 0}
cursor = self._collection.find(query_fc, projection=projection, sort=[('createdDateTime', -1)])

df_list = []
for doc in cursor:
    run_time = doc['runTime']
    price_data = doc['data']['price']
    for item in price_data:
        item['runTime'] = run_time
    df_list.append(pd.DataFrame(price_data))

# 合并所有DataFrame
df_f = pd.concat(df_list, ignore_index=True)

这样就能得到包含timeStamp、value和runTime三个字段的完整DataFrame了。

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