如何将MongoDB中多层嵌套的ObjectId转换为字符串值？

方案1：导出阶段直接处理（推荐，避免后续批量处理） #

如果还没导出数据，直接在MongoDB端用聚合管道完成转换，一次性解决所有层级的ObjectId转字符串，效率最高：

1. 针对已知嵌套层级的聚合管道
   明确嵌套结构时，用$mergeObjects替换文档中各层级的_id为字符串，数组则用$map遍历处理：

   db.collection.aggregate([
     {
       $replaceRoot: {
         newRoot: {
           $mergeObjects: [
             "$$ROOT",
             { _id: { $toString: "$_id" } },
             {
               data: {
                 $mergeObjects: [
                   "$data",
                   { _id: { $toString: "$data._id" } }
                 ]
               },
               // 处理数组类型字段的示例
               items: {
                 $map: {
                   input: "$items",
                   as: "item",
                   in: {
                     $mergeObjects: [
                       "$$item",
                       { _id: { $toString: "$$item._id" } }
                     ]
                   }
                 }
               }
             }
           ]
         }
       }
     }
     // 若要完全忽略所有_id，替换为以下$project阶段：
     // { $project: { _id: 0, "data._id": 0, "items._id": 0 } }
   ])
   

2. 针对未知嵌套层级的通用聚合管道（MongoDB 4.4+支持）
   用$function编写递归函数自动遍历所有嵌套文档和数组，统一转换ObjectId：

   db.collection.aggregate([
     {
       $addFields: {
         processed_doc: {
           $function: {
             body: function(doc) {
               function convertIds(obj) {
                 if (obj instanceof ObjectId) return obj.toString();
                 if (Array.isArray(obj)) return obj.map(convertIds);
                 if (typeof obj === 'object' && obj !== null) {
                   const newObj = {...obj};
                   for (const key in newObj) {
                     newObj[key] = convertIds(newObj[key]);
                   }
                   return newObj;
                 }
                 return obj;
               }
               return convertIds(doc);
             },
             args: ["$$ROOT"],
             lang: "js"
           }
         }
       }
     },
     { $replaceRoot: { newRoot: "$processed_doc" } }
   ])
   

3. 导出为BigQuery兼容的NDJSON格式
   用mongoexport结合上述管道导出，直接得到可导入的文件：

   mongoexport --uri "mongodb://your-host:port/your-db" --collection your-collection --aggregate '[{"$replaceRoot": {...}}]' --type json --out converted_data.ndjson
   

方案2：已导出原始数据的批量处理（Python高效版） #

如果已经拿到带ObjectId的原始数据，用Python的高效序列化工具处理，避免手动递归的性能问题：

1. 基于pymongo.json_util的处理方案
   利用官方工具解析MongoDB格式，自定义编码器将ObjectId转为字符串：

   from pymongo import json_util
   import json
   
   def encode_objectid(obj):
       if isinstance(obj, json_util.ObjectId):
           return str(obj)
       raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable")
   
   # 按行处理NDJSON格式的原始数据
   with open("raw_data.ndjson", "r") as infile, open("converted_data.ndjson", "w") as outfile:
       for line in infile:
           doc = json_util.loads(line)
           converted_line = json.dumps(doc, default=encode_objectid)
           outfile.write(converted_line + "\n")
   

2. 极致性能方案（用orjson库）
   用比标准JSON库快数倍的orjson处理百万级文档：

   import orjson
   from bson import ObjectId
   
   def default(obj):
       if isinstance(obj, ObjectId):
           return str(obj)
       raise TypeError
   
   with open("raw_data.ndjson", "rb") as infile, open("converted_data.ndjson", "wb") as outfile:
       for line in infile:
           doc = orjson.loads(line)
           converted_line = orjson.dumps(doc, default=default)
           outfile.write(converted_line + b"\n")
   

方案3：BigQuery导入时直接处理（应急方案） #

若不想提前处理数据，可通过BigQuery的UDF在导入阶段转换：

1. 创建解析MongoDB JSON的UDF

   CREATE OR REPLACE FUNCTION `your-project.your-dataset.parse_mongo_json`(json_str STRING)
   RETURNS JSON
   LANGUAGE js AS """
     function convertIds(obj) {
       if (obj && obj.$oid) return obj.$oid;
       if (Array.isArray(obj)) return obj.map(convertIds);
       if (typeof obj === 'object' && obj !== null) {
         const newObj = {...obj};
         for (const key in newObj) {
           newObj[key] = convertIds(newObj[key]);
         }
         return newObj;
       }
       return obj;
     }
     return convertIds(JSON.parse(json_str));
   """;
   

2. 导入临时表后转换为最终表

   CREATE OR REPLACE TABLE `your-project.your-dataset.final_table` AS
   SELECT 
     parse_mongo_json(raw_json) AS parsed_data
   FROM `your-project.your-dataset.temp_raw_table`;
   

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