MongoDB数据库输出JSON数据的完整指南与最佳实践帮助开发者轻松实现数据导出与交换

威震华夏关云长

1. 引言：MongoDB与JSON的关系

MongoDB是一种流行的NoSQL数据库，它使用文档存储模型，而JSON（JavaScript Object Notation）是一种轻量级的数据交换格式。这两者之间有着天然的联系，因为MongoDB中的文档本质上是BSON（Binary JSON）格式存储的，这使得JSON成为MongoDB数据导出的理想选择。

本文将全面介绍如何从MongoDB数据库中导出JSON数据，包括各种方法、工具、最佳实践以及实际应用场景，帮助开发者轻松实现数据导出与交换。

2. 为什么需要从MongoDB导出JSON数据

在实际开发中，将MongoDB数据导出为JSON格式有多种用途：

• 数据备份与迁移：将数据导出为JSON格式可以方便地进行备份或迁移到其他系统
• 数据分析：导出的JSON数据可以被各种数据分析工具处理
• 系统集成：不同系统之间通过JSON格式进行数据交换
• 开发测试：创建测试数据或模拟生产环境
• 报告生成：将数据导出后用于生成各种报告

3. MongoDB导出JSON的方法

3.1 使用mongoexport命令行工具

mongoexport是MongoDB提供的一个命令行工具，专门用于将集合中的数据导出为JSON或CSV格式。

2. mongoexport --host=localhost --port=27017 --db=mydb --collection=mycollection --out=data.json

复制代码

• --host：MongoDB服务器地址
• --port：MongoDB服务器端口
• --db：要导出的数据库名称
• --collection：要导出的集合名称
• --out：输出文件路径
• --query：导出数据的查询条件（JSON格式）
• --fields：指定要导出的字段
• --type：输出文件类型（json或csv）
• --jsonArray：将输出作为JSON数组而不是每行一个JSON对象
• --pretty：格式化JSON输出，使其更易读

1. 使用查询条件导出特定数据

2. mongoexport --db=sales --collection=orders --query='{"status": "completed", "date": {"$gte": {"$date": "2023-01-01T00:00:00Z"}}}' --out=completed_orders_2023.json

复制代码

2. 导出特定字段并格式化输出

2. mongoexport --db=users --collection=profiles --fields=name,email,age --jsonArray --pretty --out=user_profiles.json

复制代码

3. 从认证的MongoDB实例导出数据

2. mongoexport --host=cluster.example.com --port=27017 --db=app --collection=logs --username=admin --password=secret --authenticationDatabase=admin --out=logs.json

复制代码

3.2 使用MongoDB Compass图形界面

MongoDB Compass是MongoDB的官方图形界面工具，它提供了直观的方式来导出数据。

1. 连接到MongoDB服务器
2. 选择要导出的数据库和集合
3. 点击集合标签页中的”Export”按钮
4. 在弹出的对话框中选择”JSON”作为导出格式
5. 选择要导出的字段（可选）
6. 设置查询条件（可选）
7. 点击”Export”按钮并选择保存位置

• 导出整个集合：导出集合中的所有文档
• 导出查询结果：先执行查询，然后导出结果
• 字段选择：只导出选定的字段
• 输出格式：标准JSON或CSV

3.3 使用编程语言驱动程序

通过编程语言的MongoDB驱动程序，可以更灵活地控制数据导出过程。

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
4. const path = require('path');
6. async function exportToJson() {
7.   const uri = "mongodb://localhost:27017";
8.   const client = new MongoClient(uri);
9.   
10.   try {
11.     await client.connect();
12.     const database = client.db("mydb");
13.     const collection = database.collection("mycollection");
14.    
15.     // 查询条件
16.     const query = { status: "active" };
17.    
18.     // 投影（选择字段）
19.     const projection = { _id: 0, name: 1, email: 1 };
20.    
21.     // 执行查询
22.     const cursor = collection.find(query).project(projection);
23.    
24.     // 转换为数组
25.     const results = await cursor.toArray();
26.    
27.     // 转换为JSON字符串并格式化
28.     const json = JSON.stringify(results, null, 2);
29.    
30.     // 写入文件
31.     fs.writeFileSync(path.join(__dirname, 'output.json'), json);
32.    
33.     console.log(`成功导出 ${results.length} 条记录到 output.json`);
34.   } finally {
35.     await client.close();
36.   }
37. }
39. exportToJson().catch(console.error);

复制代码

2. from pymongo import MongoClient
3. import json
4. from datetime import datetime
6. def export_to_json():
7.     # 连接到MongoDB
8.     client = MongoClient('mongodb://localhost:27017/')
9.     db = client['mydb']
10.     collection = db['mycollection']
11.    
12.     # 查询条件
13.     query = {
14.         "created_at": {
15.             "$gte": datetime(2023, 1, 1),
16.             "$lt": datetime(2023, 12, 31)
17.         }
18.     }
19.    
20.     # 投影（选择字段）
21.     projection = {
22.         "_id": 0,
23.         "name": 1,
24.         "email": 1,
25.         "created_at": 1
26.     }
27.    
28.     # 执行查询
29.     cursor = collection.find(query, projection)
30.    
31.     # 转换为列表
32.     results = list(cursor)
33.    
34.     # 处理日期对象，使其可JSON序列化
35.     for doc in results:
36.         if 'created_at' in doc and isinstance(doc['created_at'], datetime):
37.             doc['created_at'] = doc['created_at'].isoformat()
38.    
39.     # 写入JSON文件
40.     with open('output.json', 'w', encoding='utf-8') as f:
41.         json.dump(results, f, ensure_ascii=False, indent=2)
42.    
43.     print(f"成功导出 {len(results)} 条记录到 output.json")
44.    
45.     # 关闭连接
46.     client.close()
48. if __name__ == "__main__":
49.     export_to_json()

复制代码

2. import com.mongodb.client.*;
3. import com.mongodb.client.model.Filters;
4. import com.mongodb.client.model.Projections;
5. import org.bson.Document;
6. import org.bson.conversions.Bson;
8. import java.io.FileWriter;
9. import java.io.IOException;
10. import java.util.ArrayList;
11. import java.util.List;
13. public class MongoToJsonExporter {
14.     public static void main(String[] args) {
15.         // 连接字符串
16.         String connectionString = "mongodb://localhost:27017";
17.         
18.         try (MongoClient mongoClient = MongoClients.create(connectionString)) {
19.             // 获取数据库和集合
20.             MongoDatabase database = mongoClient.getDatabase("mydb");
21.             MongoCollection<Document> collection = database.getCollection("mycollection");
22.             
23.             // 查询条件
24.             Bson query = Filters.eq("status", "active");
25.             
26.             // 投影（选择字段）
27.             Bson projection = Projections.fields(
28.                 Projections.excludeId(),
29.                 Projections.include("name", "email", "age")
30.             );
31.             
32.             // 执行查询
33.             FindIterable<Document> documents = collection.find(query)
34.                 .projection(projection);
35.             
36.             // 转换为List
37.             List<Document> results = new ArrayList<>();
38.             for (Document doc : documents) {
39.                 results.add(doc);
40.             }
41.             
42.             // 转换为JSON并写入文件
43.             try (FileWriter writer = new FileWriter("output.json")) {
44.                 writer.write("[\n");
45.                 for (int i = 0; i < results.size(); i++) {
46.                     writer.write("  " + results.get(i).toJson());
47.                     if (i < results.size() - 1) {
48.                         writer.write(",");
49.                     }
50.                     writer.write("\n");
51.                 }
52.                 writer.write("]");
53.             }
54.             
55.             System.out.println("成功导出 " + results.size() + " 条记录到 output.json");
56.         } catch (IOException e) {
57.             e.printStackTrace();
58.         }
59.     }
60. }

复制代码

3.4 使用聚合管道导出JSON

MongoDB的聚合管道提供了强大的数据处理能力，可以用于在导出前对数据进行转换和处理。

2. db.orders.aggregate([
3.   // 第一阶段：过滤条件
4.   { $match: { status: "completed", date: { $gte: new Date("2023-01-01") } } },
5.   
6.   // 第二阶段：选择和重命名字段
7.   { $project: {
8.       _id: 0,
9.       orderId: "$_id",
10.       customer: "$customer.name",
11.       amount: "$total",
12.       items: "$products"
13.   }},
14.   
15.   // 第三阶段：排序
16.   { $sort: { amount: -1 } }
17. ])

复制代码

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
5. async function exportWithAggregation() {
6.   const uri = "mongodb://localhost:27017";
7.   const client = new MongoClient(uri);
8.   
9.   try {
10.     await client.connect();
11.     const database = client.db("sales");
12.     const collection = database.collection("orders");
13.    
14.     // 定义聚合管道
15.     const pipeline = [
16.       // 过滤条件
17.       { $match: {
18.           status: "completed",
19.           date: { $gte: new Date("2023-01-01") }
20.       }},
21.       
22.       // 计算每个订单的项目数量
23.       { $addFields: {
24.           itemCount: { $size: "$items" }
25.       }},
26.       
27.       // 选择和重命名字段
28.       { $project: {
29.           _id: 0,
30.           orderId: "$_id",
31.           customer: 1,
32.           date: 1,
33.           total: 1,
34.           itemCount: 1
35.       }},
36.       
37.       // 排序
38.       { $sort: { total: -1 } }
39.     ];
40.    
41.     // 执行聚合
42.     const cursor = collection.aggregate(pipeline);
43.    
44.     // 转换为数组
45.     const results = await cursor.toArray();
46.    
47.     // 转换为JSON字符串并格式化
48.     const json = JSON.stringify(results, null, 2);
49.    
50.     // 写入文件
51.     fs.writeFileSync('orders_aggregated.json', json);
52.    
53.     console.log(`成功导出 ${results.length} 条记录到 orders_aggregated.json`);
54.   } finally {
55.     await client.close();
56.   }
57. }
59. exportWithAggregation().catch(console.error);

复制代码

4. 导出JSON的最佳实践

4.1 性能优化

当导出大量数据时，一次性加载所有数据可能会导致内存问题。以下是几种处理大数据集的方法：

Node.js示例：使用流式处理

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
4. const { Transform } = require('stream');
6. async function exportLargeCollection() {
7.   const uri = "mongodb://localhost:27017";
8.   const client = new MongoClient(uri);
9.   
10.   try {
11.     await client.connect();
12.     const database = client.db("large_db");
13.     const collection = database.collection("large_collection");
14.    
15.     // 创建可写流
16.     const outputStream = fs.createWriteStream('large_data.json');
17.    
18.     // 写入JSON数组的开始标记
19.     outputStream.write('[\n');
20.    
21.     let first = true;
22.     let count = 0;
23.    
24.     // 创建游标并设置批处理大小
25.     const cursor = collection.find().batchSize(1000);
26.    
27.     // 使用流式处理
28.     const transformStream = new Transform({
29.       objectMode: true,
30.       transform(doc, encoding, callback) {
31.         if (!first) {
32.           this.push(',\n');
33.         } else {
34.           first = false;
35.         }
36.         
37.         // 处理文档并转换为JSON
38.         const jsonDoc = JSON.stringify(doc, null, 2);
39.         this.push(jsonDoc);
40.         
41.         count++;
42.         if (count % 1000 === 0) {
43.           console.log(`已处理 ${count} 条记录`);
44.         }
45.         
46.         callback();
47.       }
48.     });
49.    
50.     // 管道处理
51.     cursor.pipe(transformStream).pipe(outputStream, { end: false });
52.    
53.     // 等待流完成
54.     await new Promise((resolve, reject) => {
55.       outputStream.on('finish', resolve);
56.       outputStream.on('error', reject);
57.     });
58.    
59.     // 写入JSON数组的结束标记
60.     outputStream.write('\n]');
61.     outputStream.end();
62.    
63.     console.log(`成功导出 ${count} 条记录到 large_data.json`);
64.   } finally {
65.     await client.close();
66.   }
67. }
69. exportLargeCollection().catch(console.error);

复制代码

Python示例：使用游标分批处理

2. from pymongo import MongoClient
3. import json
4. import time
6. def export_large_collection():
7.     client = MongoClient('mongodb://localhost:27017/')
8.     db = client['large_db']
9.     collection = db['large_collection']
10.    
11.     # 查询条件（无条件表示全部）
12.     query = {}
13.    
14.     # 投影（选择字段）
15.     projection = {"_id": 0}
16.    
17.     # 设置批处理大小
18.     batch_size = 1000
19.    
20.     # 打开输出文件
21.     with open('large_data.json', 'w', encoding='utf-8') as f:
22.         # 写入JSON数组的开始标记
23.         f.write('[\n')
24.         
25.         first = True
26.         count = 0
27.         
28.         # 使用游标和批处理
29.         cursor = collection.find(query, projection).batch_size(batch_size)
30.         
31.         for doc in cursor:
32.             if not first:
33.                 f.write(',\n')
34.             else:
35.                 first = False
36.             
37.             # 写入文档
38.             json.dump(doc, f, ensure_ascii=False)
39.             
40.             count += 1
41.             if count % batch_size == 0:
42.                 print(f"已处理 {count} 条记录")
43.         
44.         # 写入JSON数组的结束标记
45.         f.write('\n]')
46.    
47.     print(f"成功导出 {count} 条记录到 large_data.json")
48.     client.close()
50. if __name__ == "__main__":
51.     start_time = time.time()
52.     export_large_collection()
53.     end_time = time.time()
54.     print(f"导出完成，耗时: {end_time - start_time:.2f} 秒")

复制代码

在导出数据前，确保查询条件使用的字段有适当的索引，可以显著提高导出速度。

2. // 创建复合索引
3. db.orders.createIndex({ status: 1, date: -1 })
5. // 使用索引友好的查询
6. db.orders.find({
7.   status: "completed",
8.   date: { $gte: new Date("2023-01-01") }
9. }).sort({ date: -1 })

复制代码

4.2 数据安全

在导出数据时，可能需要处理敏感信息，如密码、个人身份信息等。

使用聚合管道屏蔽敏感字段

2. db.users.aggregate([
3.   {
4.     $project: {
5.       name: 1,
6.       email: 1,
7.       // 屏蔽密码字段
8.       password: 0,
9.       // 对电话号码进行部分屏蔽
10.       phone: {
11.         $concat: [
12.           { $substrCP: ["$phone", 0, 3] },
13.           "****",
14.           { $substrCP: ["$phone", 7, 4] }
15.         ]
16.       }
17.     }
18.   }
19. ])

复制代码

Node.js示例：在导出前处理敏感数据

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
5. async function exportWithSensitiveDataHandling() {
6.   const uri = "mongodb://localhost:27017";
7.   const client = new MongoClient(uri);
8.   
9.   try {
10.     await client.connect();
11.     const database = client.db("mydb");
12.     const collection = database.collection("users");
13.    
14.     const cursor = collection.find({});
15.    
16.     const results = [];
17.     for await (const doc of cursor) {
18.       // 创建一个新对象，只包含需要的字段
19.       const sanitizedDoc = {
20.         id: doc._id,
21.         name: doc.name,
22.         email: doc.email,
23.         // 屏蔽部分电话号码
24.         phone: doc.phone ? doc.phone.replace(/(\d{3})\d{4}(\d{4})/, '$1****$2') : null,
25.         // 屏蔽信用卡号
26.         creditCard: doc.creditCard ? '****-****-****-' + doc.creditCard.slice(-4) : null
27.       };
28.       
29.       results.push(sanitizedDoc);
30.     }
31.    
32.     // 转换为JSON字符串并格式化
33.     const json = JSON.stringify(results, null, 2);
34.    
35.     // 写入文件
36.     fs.writeFileSync('sanitized_users.json', json);
37.    
38.     console.log(`成功导出 ${results.length} 条已处理敏感信息的记录到 sanitized_users.json`);
39.   } finally {
40.     await client.close();
41.   }
42. }
44. exportWithSensitiveDataHandling().catch(console.error);

复制代码

确保只有授权用户才能导出数据，并且只导出他们有权访问的数据。

2. // 在MongoDB中设置基于角色的访问控制
3. db.createRole({
4.   role: "exportOperator",
5.   privileges: [
6.     { resource: { db: "sales", collection: "orders" }, actions: ["find"] }
7.   ],
8.   roles: []
9. })
11. // 创建用户并分配角色
12. db.createUser({
13.   user: "export_user",
14.   pwd: "secure_password",
15.   roles: [ { role: "exportOperator", db: "admin" } ]
16. })

复制代码

4.3 大数据集处理

对于分片集合，可以从每个分片并行导出数据，然后合并结果。

2. from pymongo import MongoClient
3. import json
4. import concurrent.futures
5. import os
7. def export_shard_collection():
8.     client = MongoClient('mongodb://localhost:27017/')
9.     db = client['sharded_db']
10.     collection = db['sharded_collection']
11.    
12.     # 获取分片信息
13.     shard_info = db.command("collstats", "sharded_collection")
14.     shards = client.config.shards.find()
15.    
16.     # 创建输出目录
17.     os.makedirs("sharded_exports", exist_ok=True)
18.    
19.     # 为每个分片创建导出任务
20.     def export_shard(shard):
21.         shard_host = shard['host']
22.         shard_client = MongoClient(shard_host)
23.         shard_db = shard_client['sharded_db']
24.         shard_collection = shard_db['sharded_collection']
25.         
26.         # 查询条件（可以根据分片键进行优化）
27.         query = {}
28.         
29.         # 执行查询
30.         cursor = shard_collection.find(query)
31.         
32.         # 转换为列表
33.         results = list(cursor)
34.         
35.         # 写入文件
36.         shard_name = shard['_id']
37.         output_file = f"sharded_exports/{shard_name}.json"
38.         
39.         with open(output_file, 'w', encoding='utf-8') as f:
40.             json.dump(results, f, ensure_ascii=False, indent=2)
41.         
42.         print(f"成功导出分片 {shard_name} 的 {len(results)} 条记录到 {output_file}")
43.         
44.         shard_client.close()
45.         return len(results)
46.    
47.     # 并行执行导出任务
48.     total_records = 0
49.     with concurrent.futures.ThreadPoolExecutor() as executor:
50.         futures = [executor.submit(export_shard, shard) for shard in shards]
51.         
52.         for future in concurrent.futures.as_completed(futures):
53.             count = future.result()
54.             total_records += count
55.    
56.     # 合并所有分片文件
57.     merge_shard_files("sharded_exports", "merged_output.json")
58.    
59.     print(f"所有分片导出完成，共导出 {total_records} 条记录")
60.     client.close()
62. def merge_shard_files(input_dir, output_file):
63.     import glob
64.    
65.     # 获取所有分片文件
66.     files = glob.glob(f"{input_dir}/*.json")
67.    
68.     # 创建合并后的输出文件
69.     with open(output_file, 'w', encoding='utf-8') as outfile:
70.         outfile.write('[\n')
71.         
72.         first_file = True
73.         for i, filename in enumerate(files):
74.             with open(filename, 'r', encoding='utf-8') as infile:
75.                 data = json.load(infile)
76.                
77.                 if not first_file:
78.                     outfile.write(',\n')
79.                 else:
80.                     first_file = False
81.                
82.                 # 写入每个文档
83.                 for j, doc in enumerate(data):
84.                     if j > 0:
85.                         outfile.write(',\n')
86.                     json.dump(doc, outfile, ensure_ascii=False, indent=2)
87.         
88.         outfile.write('\n]')
89.    
90.     print(f"已合并所有分片文件到 {output_file}")
92. if __name__ == "__main__":
93.     export_shard_collection()

复制代码

对于频繁变化的大型集合，可以使用增量导出策略，只导出新增或修改的数据。

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
4. const path = require('path');
6. async function incrementalExport() {
7.   const uri = "mongodb://localhost:27017";
8.   const client = new MongoClient(uri);
9.   
10.   try {
11.     await client.connect();
12.     const database = client.db("mydb");
13.     const collection = database.collection("mycollection");
14.    
15.     // 读取上次导出的时间戳
16.     let lastExportTime;
17.     try {
18.       const stats = JSON.parse(fs.readFileSync('export_stats.json', 'utf8'));
19.       lastExportTime = new Date(stats.lastExportTime);
20.     } catch (err) {
21.       // 如果不存在，使用一个较早的日期
22.       lastExportTime = new Date('2000-01-01T00:00:00Z');
23.     }
24.    
25.     console.log(`上次导出时间: ${lastExportTime.toISOString()}`);
26.    
27.     // 查询自上次导出以来新增或修改的文档
28.     const query = {
29.       $or: [
30.         { createdAt: { $gt: lastExportTime } },
31.         { updatedAt: { $gt: lastExportTime } }
32.       ]
33.     };
34.    
35.     // 执行查询
36.     const cursor = collection.find(query);
37.    
38.     // 转换为数组
39.     const results = await cursor.toArray();
40.    
41.     // 如果有新数据，追加到现有文件或创建新文件
42.     if (results.length > 0) {
43.       const now = new Date();
44.       const exportFileName = `incremental_export_${now.toISOString().replace(/:/g, '-')}.json`;
45.       const exportFilePath = path.join(__dirname, exportFileName);
46.       
47.       // 转换为JSON字符串并格式化
48.       const json = JSON.stringify(results, null, 2);
49.       
50.       // 写入文件
51.       fs.writeFileSync(exportFilePath, json);
52.       
53.       // 更新导出统计信息
54.       const exportStats = {
55.         lastExportTime: now.toISOString(),
56.         exportedCount: results.length,
57.         exportFile: exportFileName
58.       };
59.       
60.       fs.writeFileSync('export_stats.json', JSON.stringify(exportStats, null, 2));
61.       
62.       console.log(`成功导出 ${results.length} 条记录到 ${exportFileName}`);
63.     } else {
64.       console.log("没有新数据需要导出");
65.     }
66.   } finally {
67.     await client.close();
68.   }
69. }
71. incrementalExport().catch(console.error);

复制代码

5. 常见问题和解决方案

5.1 处理特殊数据类型

MongoDB支持一些JSON中不直接支持的数据类型，如ObjectId、Date、BinaryData等。在导出时需要特殊处理。

2. // 在Node.js中处理ObjectId
3. const { MongoClient, ObjectId } = require('mongodb');
4. const fs = require('fs');
6. async function exportWithObjectIdHandling() {
7.   const uri = "mongodb://localhost:27017";
8.   const client = new MongoClient(uri);
9.   
10.   try {
11.     await client.connect();
12.     const database = client.db("mydb");
13.     const collection = database.collection("mycollection");
14.    
15.     const cursor = collection.find({});
16.    
17.     const results = [];
18.     for await (const doc of cursor) {
19.       // 转换ObjectId为字符串
20.       const processedDoc = {
21.         id: doc._id.toString(),
22.         // 其他字段...
23.         data: doc.data
24.       };
25.       
26.       results.push(processedDoc);
27.     }
28.    
29.     // 转换为JSON字符串并格式化
30.     const json = JSON.stringify(results, null, 2);
31.    
32.     // 写入文件
33.     fs.writeFileSync('data_with_string_ids.json', json);
34.    
35.     console.log(`成功导出 ${results.length} 条记录，ObjectId已转换为字符串`);
36.   } finally {
37.     await client.close();
38.   }
39. }
41. exportWithObjectIdHandling().catch(console.error);

复制代码

2. from pymongo import MongoClient
3. import json
4. from datetime import datetime
6. def export_with_date_handling():
7.     client = MongoClient('mongodb://localhost:27017/')
8.     db = client['mydb']
9.     collection = db['mycollection']
10.    
11.     # 自定义JSON编码器处理日期
12.     class DateTimeEncoder(json.JSONEncoder):
13.         def default(self, obj):
14.             if isinstance(obj, datetime):
15.                 return obj.isoformat()
16.             return super().default(obj)
17.    
18.     # 执行查询
19.     cursor = collection.find({})
20.    
21.     # 转换为列表
22.     results = list(cursor)
23.    
24.     # 使用自定义编码器写入JSON文件
25.     with open('data_with_dates.json', 'w', encoding='utf-8') as f:
26.         json.dump(results, f, ensure_ascii=False, indent=2, cls=DateTimeEncoder)
27.    
28.     print(f"成功导出 {len(results)} 条记录，日期已转换为ISO格式")
29.     client.close()
31. if __name__ == "__main__":
32.     export_with_date_handling()

复制代码

5.2 处理大数据量和内存限制

当导出的数据量很大时，可能会遇到内存限制问题。以下是几种解决方案：

2. import com.mongodb.client.*;
3. import com.mongodb.client.model.Filters;
4. import org.bson.Document;
5. import org.bson.conversions.Bson;
7. import java.io.FileWriter;
8. import java.io.IOException;
9. import java.util.ArrayList;
10. import java.util.List;
12. public class BatchExportExample {
13.     public static void main(String[] args) {
14.         String connectionString = "mongodb://localhost:27017";
15.         
16.         try (MongoClient mongoClient = MongoClients.create(connectionString)) {
17.             MongoDatabase database = mongoClient.getDatabase("mydb");
18.             MongoCollection<Document> collection = database.getCollection("mycollection");
19.             
20.             // 设置批处理大小
21.             int batchSize = 1000;
22.             int totalExported = 0;
23.             
24.             try (FileWriter writer = new FileWriter("batch_export.json")) {
25.                 writer.write("[\n");
26.                
27.                 boolean firstDocument = true;
28.                
29.                 // 使用游标和批处理
30.                 MongoCursor<Document> cursor = collection.find().batchSize(batchSize).iterator();
31.                
32.                 List<Document> batch = new ArrayList<>(batchSize);
33.                
34.                 while (cursor.hasNext()) {
35.                     batch.add(cursor.next());
36.                     
37.                     if (batch.size() >= batchSize) {
38.                         // 处理当前批次
39.                         if (!firstDocument) {
40.                             writer.write(",\n");
41.                         } else {
42.                             firstDocument = false;
43.                         }
44.                         
45.                         for (int i = 0; i < batch.size(); i++) {
46.                             if (i > 0) {
47.                                 writer.write(",\n");
48.                             }
49.                             writer.write("  " + batch.get(i).toJson());
50.                         }
51.                         
52.                         totalExported += batch.size();
53.                         System.out.println("已导出 " + totalExported + " 条记录");
54.                         
55.                         // 清空批次
56.                         batch.clear();
57.                     }
58.                 }
59.                
60.                 // 处理最后一批
61.                 if (!batch.isEmpty()) {
62.                     if (!firstDocument) {
63.                         writer.write(",\n");
64.                     }
65.                     
66.                     for (int i = 0; i < batch.size(); i++) {
67.                         if (i > 0) {
68.                             writer.write(",\n");
69.                         }
70.                         writer.write("  " + batch.get(i).toJson());
71.                     }
72.                     
73.                     totalExported += batch.size();
74.                     System.out.println("已导出 " + totalExported + " 条记录");
75.                 }
76.                
77.                 writer.write("\n]");
78.             }
79.             
80.             System.out.println("导出完成，总计导出 " + totalExported + " 条记录");
81.         } catch (IOException e) {
82.             e.printStackTrace();
83.         }
84.     }
85. }

复制代码

2. from pymongo import MongoClient
3. import json
4. import concurrent.futures
5. import math
6. import time
7. from datetime import datetime
8. import os
10. def parallel_export():
11.     client = MongoClient('mongodb://localhost:27017/')
12.     db = client['mydb']
13.     collection = db['mycollection']
14.    
15.     # 获取集合中的文档总数
16.     total_docs = collection.count_documents({})
17.     print(f"集合中共有 {total_docs} 条文档")
18.    
19.     # 设置每个线程处理的文档数量
20.     docs_per_thread = 5000
21.    
22.     # 计算需要的线程数
23.     num_threads = math.ceil(total_docs / docs_per_thread)
24.     print(f"将使用 {num_threads} 个线程进行并行导出")
25.    
26.     # 创建输出目录
27.     os.makedirs("parallel_exports", exist_ok=True)
28.    
29.     # 自定义JSON编码器处理日期
30.     class DateTimeEncoder(json.JSONEncoder):
31.         def default(self, obj):
32.             if isinstance(obj, datetime):
33.                 return obj.isoformat()
34.             return super().default(obj)
35.    
36.     # 定义导出函数
37.     def export_range(start, end, thread_id):
38.         thread_client = MongoClient('mongodb://localhost:27017/')
39.         thread_db = thread_client['mydb']
40.         thread_collection = thread_db['mycollection']
41.         
42.         # 查询指定范围的文档
43.         cursor = thread_collection.find().skip(start).limit(end - start)
44.         
45.         # 转换为列表
46.         results = list(cursor)
47.         
48.         # 写入文件
49.         output_file = f"parallel_exports/thread_{thread_id}.json"
50.         with open(output_file, 'w', encoding='utf-8') as f:
51.             json.dump(results, f, ensure_ascii=False, indent=2, cls=DateTimeEncoder)
52.         
53.         thread_client.close()
54.         return len(results)
55.    
56.     # 使用线程池并行导出
57.     start_time = time.time()
58.     total_exported = 0
59.    
60.     with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
61.         futures = []
62.         
63.         for i in range(num_threads):
64.             start = i * docs_per_thread
65.             end = min((i + 1) * docs_per_thread, total_docs)
66.             
67.             future = executor.submit(export_range, start, end, i)
68.             futures.append(future)
69.         
70.         for future in concurrent.futures.as_completed(futures):
71.             count = future.result()
72.             total_exported += count
73.             print(f"线程完成，已导出 {count} 条记录，总计 {total_exported}/{total_docs}")
74.    
75.     # 合并所有线程导出的文件
76.     merge_json_files("parallel_exports", "merged_export.json")
77.    
78.     end_time = time.time()
79.     print(f"并行导出完成，共导出 {total_exported} 条记录，耗时: {end_time - start_time:.2f} 秒")
80.    
81.     client.close()
83. def merge_json_files(input_dir, output_file):
84.     import glob
85.    
86.     # 获取所有JSON文件
87.     files = glob.glob(f"{input_dir}/*.json")
88.    
89.     # 创建合并后的输出文件
90.     with open(output_file, 'w', encoding='utf-8') as outfile:
91.         outfile.write('[\n')
92.         
93.         first_file = True
94.         for i, filename in enumerate(files):
95.             with open(filename, 'r', encoding='utf-8') as infile:
96.                 data = json.load(infile)
97.                
98.                 if not first_file:
99.                     outfile.write(',\n')
100.                 else:
101.                     first_file = False
102.                
103.                 # 写入每个文档
104.                 for j, doc in enumerate(data):
105.                     if j > 0:
106.                         outfile.write(',\n')
107.                     json.dump(doc, outfile, ensure_ascii=False, indent=2)
108.         
109.         outfile.write('\n]')
110.    
111.     print(f"已合并所有文件到 {output_file}")
113. if __name__ == "__main__":
114.     parallel_export()

复制代码

5.3 处理连接和认证问题

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
5. async function exportWithAuth() {
6.   // 使用连接字符串处理认证
7.   const uri = "mongodb://username:password@cluster.example.com:27017,cluster.example.com:27018,cluster.example.com:27019/mydb?replicaSet=myReplicaSet&authSource=admin";
8.   
9.   const client = new MongoClient(uri, {
10.     // 连接池选项
11.     maxPoolSize: 10,
12.     minPoolSize: 1,
13.     maxIdleTimeMS: 30000,
14.     waitQueueTimeoutMS: 5000,
15.    
16.     // 重试选项
17.     connectTimeoutMS: 10000,
18.     socketTimeoutMS: 45000,
19.     retryWrites: true,
20.    
21.     // SSL选项（如果需要）
22.     ssl: true,
23.     sslValidate: true,
24.     sslCA: process.env.SSL_CA_FILE
25.   });
26.   
27.   try {
28.     await client.connect();
29.     const database = client.db("mydb");
30.     const collection = database.collection("mycollection");
31.    
32.     // 执行查询
33.     const cursor = collection.find({});
34.    
35.     // 转换为数组
36.     const results = await cursor.toArray();
37.    
38.     // 转换为JSON字符串并格式化
39.     const json = JSON.stringify(results, null, 2);
40.    
41.     // 写入文件
42.     fs.writeFileSync('auth_export.json', json);
43.    
44.     console.log(`成功导出 ${results.length} 条记录`);
45.   } catch (err) {
46.     console.error("导出过程中出错:", err);
47.   } finally {
48.     await client.close();
49.   }
50. }
52. exportWithAuth().catch(console.error);

复制代码

2. from pymongo import MongoClient
3. import paramiko
4. import json
5. import sshtunnel
6. from datetime import datetime
8. def export_via_ssh_tunnel():
9.     # SSH隧道配置
10.     ssh_host = 'ssh.example.com'
11.     ssh_port = 22
12.     ssh_username = 'ssh_user'
13.     ssh_pkey = paramiko.RSAKey.from_private_key_file('/path/to/private/key')
14.    
15.     # MongoDB配置
16.     mongo_host = 'localhost'  # 相对于SSH服务器
17.     mongo_port = 27017
18.     mongo_db = 'mydb'
19.     mongo_collection = 'mycollection'
20.    
21.     # 创建SSH隧道
22.     server = sshtunnel.SSHTunnelForwarder(
23.         (ssh_host, ssh_port),
24.         ssh_username=ssh_username,
25.         ssh_pkey=ssh_pkey,
26.         remote_bind_address=(mongo_host, mongo_port)
27.     )
28.    
29.     try:
30.         # 启动SSH隧道
31.         server.start()
32.         
33.         # 通过隧道连接到MongoDB
34.         client = MongoClient(
35.             host='localhost',
36.             port=server.local_bind_port,
37.             username='mongo_user',
38.             password='mongo_password',
39.             authSource='admin'
40.         )
41.         
42.         db = client[mongo_db]
43.         collection = db[mongo_collection]
44.         
45.         # 自定义JSON编码器处理日期
46.         class DateTimeEncoder(json.JSONEncoder):
47.             def default(self, obj):
48.                 if isinstance(obj, datetime):
49.                     return obj.isoformat()
50.                 return super().default(obj)
51.         
52.         # 执行查询
53.         cursor = collection.find({})
54.         
55.         # 转换为列表
56.         results = list(cursor)
57.         
58.         # 写入JSON文件
59.         with open('ssh_tunnel_export.json', 'w', encoding='utf-8') as f:
60.             json.dump(results, f, ensure_ascii=False, indent=2, cls=DateTimeEncoder)
61.         
62.         print(f"成功导出 {len(results)} 条记录")
63.         
64.     except Exception as e:
65.         print(f"导出过程中出错: {e}")
66.     finally:
67.         # 关闭连接和隧道
68.         if 'client' in locals():
69.             client.close()
70.         server.stop()
72. if __name__ == "__main__":
73.     export_via_ssh_tunnel()

复制代码

6. 实际应用场景和案例研究

6.1 数据备份与恢复

2. #!/bin/bash
4. # MongoDB备份脚本
5. # 配置变量
6. MONGO_HOST="localhost"
7. MONGO_PORT="27017"
8. MONGO_USER="backup_user"
9. MONGO_PASS="backup_password"
10. MONGO_AUTH_DB="admin"
11. BACKUP_DIR="/var/backups/mongodb"
12. DATE=$(date +%Y%m%d_%H%M%S)
13. RETENTION_DAYS=7
15. # 创建备份目录
16. mkdir -p "$BACKUP_DIR/$DATE"
18. # 获取所有数据库列表
19. DATABASES=$(mongo --host $MONGO_HOST --port $MONGO_PORT -u $MONGO_USER -p $MONGO_PASS --authenticationDatabase $MONGO_AUTH_DB --quiet --eval "db.getMongo().getDBNames().join(' ')")
21. # 备份每个数据库
22. for DB in $DATABASES; do
23.     echo "备份数据库: $DB"
24.    
25.     # 获取数据库中的所有集合
26.     COLLECTIONS=$(mongo --host $MONGO_HOST --port $MONGO_PORT -u $MONGO_USER -p $MONGO_PASS --authenticationDatabase $MONGO_AUTH_DB $DB --quiet --eval "db.getCollectionNames().join(' ')")
27.    
28.     # 为每个集合创建备份目录
29.     mkdir -p "$BACKUP_DIR/$DATE/$DB"
30.    
31.     # 备份每个集合
32.     for COLLECTION in $COLLECTIONS; do
33.         echo "  备份集合: $DB.$COLLECTION"
34.         mongoexport --host $MONGO_HOST --port $MONGO_PORT -u $MONGO_USER -p $MONGO_PASS --authenticationDatabase $MONGO_AUTH_DB \
35.             --db $DB --collection $COLLECTION \
36.             --out "$BACKUP_DIR/$DATE/$DB/$COLLECTION.json" \
37.             --jsonArray
38.     done
39. done
41. # 创建备份清单
42. echo "备份时间: $DATE" > "$BACKUP_DIR/$DATE/backup_info.txt"
43. echo "数据库列表: $DATABASES" >> "$BACKUP_DIR/$DATE/backup_info.txt"
45. # 压缩备份
46. echo "压缩备份文件..."
47. tar -czf "$BACKUP_DIR/mongodb_backup_$DATE.tar.gz" -C "$BACKUP_DIR" "$DATE"
49. # 删除未压缩的备份
50. rm -rf "$BACKUP_DIR/$DATE"
52. # 清理旧备份
53. echo "清理超过 $RETENTION_DAYS 天的旧备份..."
54. find "$BACKUP_DIR" -name "mongodb_backup_*.tar.gz" -type f -mtime +$RETENTION_DAYS -delete
56. echo "备份完成: $BACKUP_DIR/mongodb_backup_$DATE.tar.gz"

复制代码

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
4. const path = require('path');
6. async function restoreFromBackup(backupDir) {
7.   const uri = "mongodb://localhost:27017";
8.   const client = new MongoClient(uri);
9.   
10.   try {
11.     await client.connect();
12.    
13.     // 读取备份目录中的所有文件
14.     const backupFiles = fs.readdirSync(backupDir, { withFileTypes: true });
15.    
16.     for (const file of backupFiles) {
17.       if (file.isDirectory()) {
18.         // 处理数据库目录
19.         const dbName = file.name;
20.         const db = client.db(dbName);
21.         
22.         console.log(`恢复数据库: ${dbName}`);
23.         
24.         // 读取数据库中的所有集合文件
25.         const collectionFiles = fs.readdirSync(path.join(backupDir, dbName));
26.         
27.         for (const collectionFile of collectionFiles) {
28.           if (collectionFile.endsWith('.json')) {
29.             // 处理集合文件
30.             const collectionName = path.basename(collectionFile, '.json');
31.             const collection = db.collection(collectionName);
32.             
33.             console.log(`  恢复集合: ${dbName}.${collectionName}`);
34.             
35.             // 读取JSON文件
36.             const jsonContent = fs.readFileSync(path.join(backupDir, dbName, collectionFile), 'utf8');
37.             const documents = JSON.parse(jsonContent);
38.             
39.             if (documents.length > 0) {
40.               // 清空现有集合（可选）
41.               await collection.deleteMany({});
42.               
43.               // 插入文档
44.               await collection.insertMany(documents);
45.               
46.               console.log(`    恢复了 ${documents.length} 条文档`);
47.             }
48.           }
49.         }
50.       }
51.     }
52.    
53.     console.log("备份恢复完成");
54.   } finally {
55.     await client.close();
56.   }
57. }
59. // 使用示例
60. // restoreFromBackup('/var/backups/mongodb/20231115_143022').catch(console.error);

复制代码

6.2 数据迁移

2. import pymongo
3. import json
4. import psycopg2
5. from datetime import datetime
6. import sys
8. def mongo_to_postgres():
9.     # MongoDB连接配置
10.     mongo_client = pymongo.MongoClient("mongodb://localhost:27017/")
11.     mongo_db = mongo_client["source_db"]
12.    
13.     # PostgreSQL连接配置
14.     pg_conn = psycopg2.connect(
15.         host="localhost",
16.         database="target_db",
17.         user="postgres",
18.         password="postgres"
19.     )
20.     pg_cursor = pg_conn.cursor()
21.    
22.     try:
23.         # 迁移用户数据
24.         print("迁移用户数据...")
25.         mongo_users = mongo_db["users"].find({})
26.         
27.         # 创建PostgreSQL表（如果不存在）
28.         pg_cursor.execute("""
29.             CREATE TABLE IF NOT EXISTS users (
30.                 id SERIAL PRIMARY KEY,
31.                 mongo_id VARCHAR(24) UNIQUE,
32.                 name VARCHAR(100) NOT NULL,
33.                 email VARCHAR(100) UNIQUE,
34.                 age INTEGER,
35.                 created_at TIMESTAMP,
36.                 updated_at TIMESTAMP
37.             )
38.         """)
39.         
40.         # 插入用户数据
41.         for user in mongo_users:
42.             pg_cursor.execute(
43.                 """
44.                 INSERT INTO users (mongo_id, name, email, age, created_at, updated_at)
45.                 VALUES (%s, %s, %s, %s, %s, %s)
46.                 ON CONFLICT (mongo_id) DO UPDATE SET
47.                     name = EXCLUDED.name,
48.                     email = EXCLUDED.email,
49.                     age = EXCLUDED.age,
50.                     updated_at = EXCLUDED.updated_at
51.                 """,
52.                 (
53.                     str(user["_id"]),
54.                     user.get("name"),
55.                     user.get("email"),
56.                     user.get("age"),
57.                     user.get("created_at"),
58.                     user.get("updated_at")
59.                 )
60.             )
61.         
62.         pg_conn.commit()
63.         print(f"用户数据迁移完成，共迁移 {mongo_db['users'].count_documents({})} 条记录")
64.         
65.         # 迁移订单数据
66.         print("迁移订单数据...")
67.         mongo_orders = mongo_db["orders"].find({})
68.         
69.         # 创建PostgreSQL表（如果不存在）
70.         pg_cursor.execute("""
71.             CREATE TABLE IF NOT EXISTS orders (
72.                 id SERIAL PRIMARY KEY,
73.                 mongo_id VARCHAR(24) UNIQUE,
74.                 user_id INTEGER REFERENCES users(id),
75.                 order_number VARCHAR(50),
76.                 total DECIMAL(10,2),
77.                 status VARCHAR(20),
78.                 order_date TIMESTAMP,
79.                 created_at TIMESTAMP,
80.                 updated_at TIMESTAMP
81.             )
82.         """)
83.         
84.         # 创建用户ID映射（MongoDB _id 到 PostgreSQL id）
85.         pg_cursor.execute("SELECT mongo_id, id FROM users")
86.         user_id_map = {row[0]: row[1] for row in pg_cursor.fetchall()}
87.         
88.         # 插入订单数据
89.         for order in mongo_orders:
90.             # 获取对应的PostgreSQL用户ID
91.             user_mongo_id = str(order.get("user_id"))
92.             pg_user_id = user_id_map.get(user_mongo_id)
93.             
94.             if pg_user_id is None:
95.                 print(f"警告: 找不到用户ID {user_mongo_id}，跳过订单 {order.get('_id')}")
96.                 continue
97.             
98.             pg_cursor.execute(
99.                 """
100.                 INSERT INTO orders (mongo_id, user_id, order_number, total, status, order_date, created_at, updated_at)
101.                 VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
102.                 ON CONFLICT (mongo_id) DO UPDATE SET
103.                     user_id = EXCLUDED.user_id,
104.                     order_number = EXCLUDED.order_number,
105.                     total = EXCLUDED.total,
106.                     status = EXCLUDED.status,
107.                     order_date = EXCLUDED.order_date,
108.                     updated_at = EXCLUDED.updated_at
109.                 """,
110.                 (
111.                     str(order["_id"]),
112.                     pg_user_id,
113.                     order.get("order_number"),
114.                     order.get("total"),
115.                     order.get("status"),
116.                     order.get("order_date"),
117.                     order.get("created_at"),
118.                     order.get("updated_at")
119.                 )
120.             )
121.         
122.         pg_conn.commit()
123.         print(f"订单数据迁移完成，共迁移 {mongo_db['orders'].count_documents({})} 条记录")
124.         
125.         print("数据迁移完成")
126.         
127.     except Exception as e:
128.         print(f"迁移过程中出错: {e}")
129.         pg_conn.rollback()
130.     finally:
131.         pg_cursor.close()
132.         pg_conn.close()
133.         mongo_client.close()
135. if __name__ == "__main__":
136.     mongo_to_postgres()

复制代码

2. const { MongoClient } = require('mongodb');
3. const mysql = require('mysql2/promise');
4. const fs = require('fs');
5. const path = require('path');
7. async function mysqlToMongoDB() {
8.   // MySQL连接配置
9.   const mysqlConnection = await mysql.createConnection({
10.     host: 'localhost',
11.     user: 'root',
12.     password: 'password',
13.     database: 'source_db'
14.   });
15.   
16.   // MongoDB连接配置
17.   const mongoUri = "mongodb://localhost:27017";
18.   const mongoClient = new MongoClient(mongoUri);
19.   
20.   try {
21.     await mongoClient.connect();
22.     const mongoDb = mongoClient.db('target_db');
23.    
24.     // 迁移用户数据
25.     console.log('迁移用户数据...');
26.     const [users] = await mysqlConnection.execute('SELECT * FROM users');
27.     const usersCollection = mongoDb.collection('users');
28.    
29.     // 转换用户数据格式
30.     const userDocuments = users.map(user => ({
31.       _id: user.id,  // 使用MySQL ID作为MongoDB _id
32.       name: user.name,
33.       email: user.email,
34.       age: user.age,
35.       profile: {
36.         address: user.address,
37.         phone: user.phone
38.       },
39.       createdAt: user.created_at,
40.       updatedAt: user.updated_at
41.     }));
42.    
43.     // 插入用户数据
44.     if (userDocuments.length > 0) {
45.       await usersCollection.insertMany(userDocuments);
46.       console.log(`用户数据迁移完成，共迁移 ${userDocuments.length} 条记录`);
47.     }
48.    
49.     // 迁移产品数据
50.     console.log('迁移产品数据...');
51.     const [products] = await mysqlConnection.execute('SELECT * FROM products');
52.     const productsCollection = mongoDb.collection('products');
53.    
54.     // 转换产品数据格式
55.     const productDocuments = products.map(product => ({
56.       _id: product.id,
57.       name: product.name,
58.       description: product.description,
59.       price: parseFloat(product.price),
60.       category: product.category,
61.       tags: product.tags ? product.tags.split(',') : [],
62.       specifications: JSON.parse(product.specifications || '{}'),
63.       stock: {
64.         quantity: product.stock_quantity,
65.         reserved: product.stock_reserved || 0,
66.         available: product.stock_quantity - (product.stock_reserved || 0)
67.       },
68.       createdAt: product.created_at,
69.       updatedAt: product.updated_at
70.     }));
71.    
72.     // 插入产品数据
73.     if (productDocuments.length > 0) {
74.       await productsCollection.insertMany(productDocuments);
75.       console.log(`产品数据迁移完成，共迁移 ${productDocuments.length} 条记录`);
76.     }
77.    
78.     // 迁移订单数据
79.     console.log('迁移订单数据...');
80.     const [orders] = await mysqlConnection.execute('SELECT * FROM orders');
81.     const ordersCollection = mongoDb.collection('orders');
82.    
83.     // 获取订单项数据
84.     const [orderItems] = await mysqlConnection.execute('SELECT * FROM order_items');
85.    
86.     // 按订单ID分组订单项
87.     const orderItemsByOrderId = {};
88.     orderItems.forEach(item => {
89.       if (!orderItemsByOrderId[item.order_id]) {
90.         orderItemsByOrderId[item.order_id] = [];
91.       }
92.       orderItemsByOrderId[item.order_id].push({
93.         productId: item.product_id,
94.         quantity: item.quantity,
95.         price: parseFloat(item.price),
96.         subtotal: parseFloat(item.subtotal)
97.       });
98.     });
99.    
100.     // 转换订单数据格式
101.     const orderDocuments = orders.map(order => {
102.       const items = orderItemsByOrderId[order.id] || [];
103.       
104.       return {
105.         _id: order.id,
106.         userId: order.user_id,
107.         orderNumber: order.order_number,
108.         status: order.status,
109.         items: items,
110.         subtotal: parseFloat(order.subtotal),
111.         tax: parseFloat(order.tax),
112.         shipping: parseFloat(order.shipping),
113.         total: parseFloat(order.total),
114.         shippingAddress: {
115.           street: order.shipping_street,
116.           city: order.shipping_city,
117.           state: order.shipping_state,
118.           zipCode: order.shipping_zip_code,
119.           country: order.shipping_country
120.         },
121.         paymentMethod: order.payment_method,
122.         paymentStatus: order.payment_status,
123.         orderDate: order.order_date,
124.         createdAt: order.created_at,
125.         updatedAt: order.updated_at
126.       };
127.     });
128.    
129.     // 插入订单数据
130.     if (orderDocuments.length > 0) {
131.       await ordersCollection.insertMany(orderDocuments);
132.       console.log(`订单数据迁移完成，共迁移 ${orderDocuments.length} 条记录`);
133.     }
134.    
135.     console.log('数据迁移完成');
136.    
137.   } catch (error) {
138.     console.error('迁移过程中出错:', error);
139.   } finally {
140.     await mysqlConnection.end();
141.     await mongoClient.close();
142.   }
143. }
145. mysqlToMongoDB().catch(console.error);

复制代码

6.3 数据分析和报告生成

2. from pymongo import MongoClient
3. import json
4. import pandas as pd
5. from datetime import datetime, timedelta
6. import matplotlib.pyplot as plt
7. import seaborn as sns
9. def sales_data_analysis():
10.     # 连接到MongoDB
11.     client = MongoClient('mongodb://localhost:27017/')
12.     db = client['sales_db']
13.    
14.     # 定义分析日期范围
15.     end_date = datetime.now()
16.     start_date = end_date - timedelta(days=90)  # 最近90天
17.    
18.     # 查询销售数据
19.     pipeline = [
20.         {
21.             '$match': {
22.                 'order_date': {
23.                     '$gte': start_date,
24.                     '$lte': end_date
25.                 },
26.                 'status': 'completed'
27.             }
28.         },
29.         {
30.             '$project': {
31.                 '_id': 0,
32.                 'order_id': '$_id',
33.                 'customer_id': 1,
34.                 'order_date': 1,
35.                 'total': 1,
36.                 'items': 1,
37.                 'payment_method': 1,
38.                 'year': {'$year': '$order_date'},
39.                 'month': {'$month': '$order_date'},
40.                 'day': {'$dayOfMonth': '$order_date'},
41.                 'day_of_week': {'$dayOfWeek': '$order_date'}
42.             }
43.         }
44.     ]
45.    
46.     # 执行聚合查询
47.     sales_data = list(db.orders.aggregate(pipeline))
48.    
49.     # 导出原始数据到JSON
50.     with open('sales_data_raw.json', 'w', encoding='utf-8') as f:
51.         json.dump(sales_data, f, ensure_ascii=False, indent=2, default=str)
52.    
53.     print(f"已导出 {len(sales_data)} 条销售记录到 sales_data_raw.json")
54.    
55.     # 转换为DataFrame进行分析
56.     df = pd.DataFrame(sales_data)
57.    
58.     # 1. 按日期分析销售趋势
59.     df['order_date'] = pd.to_datetime(df['order_date'])
60.     df.set_index('order_date', inplace=True)
61.    
62.     # 按天汇总销售数据
63.     daily_sales = df['total'].resample('D').sum()
64.    
65.     # 导出每日销售数据
66.     daily_sales.to_json('daily_sales.json', date_format='iso')
67.    
68.     # 2. 按产品分析销售情况
69.     product_sales = {}
70.     for _, order in df.iterrows():
71.         for item in order['items']:
72.             product_id = item['product_id']
73.             if product_id not in product_sales:
74.                 product_sales[product_id] = {
75.                     'quantity': 0,
76.                     'revenue': 0
77.                 }
78.             product_sales[product_id]['quantity'] += item['quantity']
79.             product_sales[product_id]['revenue'] += item['price'] * item['quantity']
80.    
81.     # 转换为DataFrame并排序
82.     product_df = pd.DataFrame.from_dict(product_sales, orient='index')
83.     product_df = product_df.sort_values('revenue', ascending=False)
84.    
85.     # 导出产品销售数据
86.     product_df.to_json('product_sales.json', orient='index')
87.    
88.     # 3. 按支付方式分析
89.     payment_stats = df.groupby('payment_method')['total'].agg(['count', 'sum', 'mean'])
90.    
91.     # 导出支付方式统计数据
92.     payment_stats.to_json('payment_stats.json', orient='index')
93.    
94.     # 4. 按星期几分析销售模式
95.     weekday_stats = df.groupby('day_of_week')['total'].agg(['count', 'sum', 'mean'])
96.    
97.     # 导出星期几统计数据
98.     weekday_stats.to_json('weekday_stats.json', orient='index')
99.    
100.     # 生成可视化图表
101.     plt.figure(figsize=(15, 10))
102.    
103.     # 每日销售趋势图
104.     plt.subplot(2, 2, 1)
105.     daily_sales.plot()
106.     plt.title('每日销售趋势')
107.     plt.xlabel('日期')
108.     plt.ylabel('销售额')
109.     plt.grid(True)
110.    
111.     # 产品销售Top 10
112.     plt.subplot(2, 2, 2)
113.     top_products = product_df.head(10)
114.     top_products['revenue'].plot(kind='bar')
115.     plt.title('产品销售额Top 10')
116.     plt.xlabel('产品ID')
117.     plt.ylabel('销售额')
118.     plt.xticks(rotation=45)
119.    
120.     # 支付方式分布
121.     plt.subplot(2, 2, 3)
122.     payment_stats['count'].plot(kind='pie', autopct='%1.1f%%')
123.     plt.title('支付方式分布')
124.     plt.ylabel('')
125.    
126.     # 星期几销售模式
127.     plt.subplot(2, 2, 4)
128.     weekday_names = ['周日', '周一', '周二', '周三', '周四', '周五', '周六']
129.     weekday_stats.index = [weekday_names[i-1] for i in weekday_stats.index]
130.     weekday_stats['sum'].plot(kind='bar')
131.     plt.title('按星期几的销售额')
132.     plt.xlabel('星期')
133.     plt.ylabel('销售额')
134.    
135.     plt.tight_layout()
136.     plt.savefig('sales_analysis.png')
137.     plt.close()
138.    
139.     # 生成分析报告
140.     report = f"""
141. # 销售数据分析报告
143. ## 分析期间
144. {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}
146. ## 总体销售情况
147. - 总订单数: {len(df)}
148. - 总销售额: {df['total'].sum():.2f}
149. - 平均订单金额: {df['total'].mean():.2f}
151. ## 销售趋势
152. - 最高单日销售额: {daily_sales.max():.2f} ({daily_sales.idxmax().strftime('%Y-%m-%d')})
153. - 最低单日销售额: {daily_sales.min():.2f} ({daily_sales.idxmin().strftime('%Y-%m-%d')})
154. - 平均每日销售额: {daily_sales.mean():.2f}
156. ## 热销产品
157. - 销售额最高的产品: {product_df.index[0]} (销售额: {product_df.iloc[0]['revenue']:.2f})
158. - 销量最高的产品: {product_df.sort_values('quantity', ascending=False).index[0]} (销量: {product_df.sort_values('quantity', ascending=False).iloc[0]['quantity']})
160. ## 支付方式分析
161. """
162.    
163.     for payment_method, stats in payment_stats.iterrows():
164.         report += f"- {payment_method}: {stats['count']} 笔订单, 总额 {stats['sum']:.2f}, 平均 {stats['mean']:.2f}\n"
165.    
166.     report += "\n## 星期几销售模式\n"
167.    
168.     for day, stats in weekday_stats.iterrows():
169.         report += f"- {day}: {stats['count']} 笔订单, 总额 {stats['sum']:.2f}, 平均 {stats['mean']:.2f}\n"
170.    
171.     # 保存报告
172.     with open('sales_analysis_report.md', 'w', encoding='utf-8') as f:
173.         f.write(report)
174.    
175.     print("销售数据分析完成，已生成以下文件:")
176.     print("- sales_data_raw.json: 原始销售数据")
177.     print("- daily_sales.json: 每日销售数据")
178.     print("- product_sales.json: 产品销售数据")
179.     print("- payment_stats.json: 支付方式统计数据")
180.     print("- weekday_stats.json: 星期几销售数据")
181.     print("- sales_analysis.png: 销售分析图表")
182.     print("- sales_analysis_report.md: 销售分析报告")
183.    
184.     client.close()
186. if __name__ == "__main__":
187.     sales_data_analysis()

复制代码

2. const { MongoClient } = require('mongodb');
3. const fs = require('fs');
5. async function userBehaviorAnalysis() {
6.   const uri = "mongodb://localhost:27017";
7.   const client = new MongoClient(uri);
8.   
9.   try {
10.     await client.connect();
11.     const database = client.db("analytics_db");
12.    
13.     // 1. 用户活动分析
14.     console.log("分析用户活动...");
15.    
16.     const userActivityPipeline = [
17.       {
18.         $match: {
19.           timestamp: {
20.             $gte: new Date(new Date() - 30 * 24 * 60 * 60 * 1000) // 最近30天
21.           }
22.         }
23.       },
24.       {
25.         $group: {
26.           _id: "$userId",
27.           sessionCount: { $sum: 1 },
28.           lastActivity: { $max: "$timestamp" },
29.           activities: { $push: "$action" }
30.         }
31.       },
32.       {
33.         $project: {
34.           userId: "$_id",
35.           sessionCount: 1,
36.           lastActivity: 1,
37.           uniqueActivities: { $size: { $setUnion: "$activities" } },
38.           mostCommonActivity: {
39.             $arrayElemAt: [
40.               {
41.                 $slice: [
42.                   {
43.                     $sortArray: {
44.                       input: {
45.                         $map: {
46.                           input: { $setUnion: "$activities" },
47.                           as: "activity",
48.                           in: {
49.                             activity: "$$activity",
50.                             count: {
51.                               $size: {
52.                                 $filter: {
53.                                   input: "$activities",
54.                                   cond: { $eq: ["$$this", "$$activity"] }
55.                                 }
56.                               }
57.                             }
58.                           }
59.                         }
60.                       },
61.                       sortBy: { count: -1 }
62.                     }
63.                   },
64.                   1
65.                 ]
66.               },
67.               0
68.             ]
69.           }
70.         }
71.       },
72.       {
73.         $sort: { sessionCount: -1 }
74.       }
75.     ];
76.    
77.     const userActivity = await database.collection("user_sessions").aggregate(userActivityPipeline).toArray();
78.    
79.     // 导出用户活动数据
80.     fs.writeFileSync('user_activity_analysis.json', JSON.stringify(userActivity, null, 2));
81.     console.log(`已导出 ${userActivity.length} 个用户的活动分析数据`);
82.    
83.     // 2. 用户路径分析
84.     console.log("分析用户路径...");
85.    
86.     const userPathPipeline = [
87.       {
88.         $match: {
89.           timestamp: {
90.             $gte: new Date(new Date() - 30 * 24 * 60 * 60 * 1000) // 最近30天
91.           }
92.         }
93.       },
94.       {
95.         $sort: { userId: 1, timestamp: 1 }
96.       },
97.       {
98.         $group: {
99.           _id: "$userId",
100.           paths: {
101.             $push: {
102.               page: "$page",
103.               timestamp: "$timestamp"
104.             }
105.           }
106.         }
107.       },
108.       {
109.         $project: {
110.           userId: "$_id",
111.           pathSequence: {
112.             $map: {
113.               input: "$paths",
114.               as: "path",
115.               in: "$$path.page"
116.             }
117.           },
118.           sessionCount: { $size: "$paths" },
119.           timeSpent: {
120.             $reduce: {
121.               input: { $slice: ["$paths", 1, { $size: "$paths" }] },
122.               initialValue: 0,
123.               in: {
124.                 $add: [
125.                   "$$value",
126.                   {
127.                     $subtract: [
128.                       "$$this.timestamp",
129.                       {
130.                         $arrayElemAt: [
131.                           "$paths",
132.                           { $subtract: [{ $indexOfArray: ["$paths", "$$this"] }, 1] }
133.                         ]
134.                       }.timestamp
135.                     ]
136.                   }
137.                 ]
138.               }
139.             }
140.           }
141.         }
142.       },
143.       {
144.         $addFields: {
145.           avgTimePerPage: {
146.             $cond: {
147.               if: { $gt: [{ $size: "$pathSequence" }, 1] },
148.               then: { $divide: ["$timeSpent", { $subtract: [{ $size: "$pathSequence" }, 1] } },
149.               else: 0
150.             }
151.           }
152.         }
153.       },
154.       {
155.         $sort: { sessionCount: -1 }
156.       }
157.     ];
158.    
159.     const userPaths = await database.collection("page_views").aggregate(userPathPipeline).toArray();
160.    
161.     // 导出用户路径数据
162.     fs.writeFileSync('user_path_analysis.json', JSON.stringify(userPaths, null, 2));
163.     console.log(`已导出 ${userPaths.length} 个用户的路径分析数据`);
164.    
165.     // 3. 转化漏斗分析
166.     console.log("分析转化漏斗...");
167.    
168.     const conversionFunnelPipeline = [
169.       {
170.         $match: {
171.           timestamp: {
172.             $gte: new Date(new Date() - 30 * 24 * 60 * 60 * 1000) // 最近30天
173.           }
174.         }
175.       },
176.       {
177.         $group: {
178.           _id: "$userId",
179.           events: { $push: "$event" },
180.           firstEvent: { $first: "$event" },
181.           lastEvent: { $last: "$event" },
182.           uniqueEvents: { $addToSet: "$event" }
183.         }
184.       },
185.       {
186.         $project: {
187.           userId: "$_id",
188.           events: 1,
189.           firstEvent: 1,
190.           lastEvent: 1,
191.           visitedHomepage: { $in: ["homepage", "$uniqueEvents"] },
192.           viewedProduct: { $in: ["view_product", "$uniqueEvents"] },
193.           addedToCart: { $in: ["add_to_cart", "$uniqueEvents"] },
194.           initiatedCheckout: { $in: ["initiate_checkout", "$uniqueEvents"] },
195.           completedPurchase: { $in: ["purchase", "$uniqueEvents"] }
196.         }
197.       },
198.       {
199.         $group: {
200.           _id: null,
201.           totalUsers: { $sum: 1 },
202.           homepageVisits: { $sum: { $cond: ["$visitedHomepage", 1, 0] } },
203.           productViews: { $sum: { $cond: ["$viewedProduct", 1, 0] } },
204.           cartAdditions: { $sum: { $cond: ["$addedToCart", 1, 0] } },
205.           checkouts: { $sum: { $cond: ["$initiatedCheckout", 1, 0] } },
206.           purchases: { $sum: { $cond: ["$completedPurchase", 1, 0] } }
207.         }
208.       },
209.       {
210.         $project: {
211.           _id: 0,
212.           totalUsers: 1,
213.           homepageVisits: 1,
214.           productViews: 1,
215.           cartAdditions: 1,
216.           checkouts: 1,
217.           purchases: 1,
218.           homepageConversionRate: { $divide: ["$homepageVisits", "$totalUsers"] },
219.           productViewRate: { $divide: ["$productViews", "$homepageVisits"] },
220.           cartAdditionRate: { $divide: ["$cartAdditions", "$productViews"] },
221.           checkoutRate: { $divide: ["$checkouts", "$cartAdditions"] },
222.           purchaseRate: { $divide: ["$purchases", "$checkouts"] },
223.           overallConversionRate: { $divide: ["$purchases", "$totalUsers"] }
224.         }
225.       }
226.     ];
227.    
228.     const conversionFunnel = await database.collection("user_events").aggregate(conversionFunnelPipeline).toArray();
229.    
230.     // 导出转化漏斗数据
231.     fs.writeFileSync('conversion_funnel_analysis.json', JSON.stringify(conversionFunnel, null, 2));
232.     console.log("已导出转化漏斗分析数据");
233.    
234.     // 4. 用户留存分析
235.     console.log("分析用户留存...");
236.    
237.     const userRetentionPipeline = [
238.       {
239.         $match: {
240.           event: "login",
241.           timestamp: {
242.             $gte: new Date(new Date() - 90 * 24 * 60 * 60 * 1000) // 最近90天
243.           }
244.         }
245.       },
246.       {
247.         $group: {
248.           _id: "$userId",
249.           firstLogin: { $min: "$timestamp" },
250.           lastLogin: { $max: "$timestamp" },
251.           loginCount: { $sum: 1 },
252.           loginDays: {
253.             $addToSet: {
254.               $dateToString: {
255.                 format: "%Y-%m-%d",
256.                 date: "$timestamp"
257.               }
258.             }
259.           }
260.         }
261.       },
262.       {
263.         $project: {
264.           userId: "$_id",
265.           firstLogin: 1,
266.           lastLogin: 1,
267.           loginCount: 1,
268.           activeDays: { $size: "$loginDays" },
269.           cohort: {
270.             $dateToString: {
271.               format: "%Y-%m",
272.               date: "$firstLogin"
273.             }
274.           }
275.         }
276.       },
277.       {
278.         $group: {
279.           _id: "$cohort",
280.           users: { $push: "$$ROOT" },
281.           userCount: { $sum: 1 }
282.         }
283.       },
284.       {
285.         $project: {
286.           cohort: "$_id",
287.           userCount: 1,
288.           users: 1,
289.           day1Retention: {
290.             $divide: [
291.               {
292.                 $size: {
293.                   $filter: {
294.                     input: "$users",
295.                     cond: {
296.                       $gte: [
297.                         {
298.                           $dateDiff: {
299.                             startDate: "$$this.firstLogin",
300.                             endDate: "$$this.lastLogin",
301.                             unit: "day"
302.                           }
303.                         },
304.                         1
305.                       ]
306.                     }
307.                   }
308.                 }
309.               },
310.               "$userCount"
311.             ]
312.           },
313.           day7Retention: {
314.             $divide: [
315.               {
316.                 $size: {
317.                   $filter: {
318.                     input: "$users",
319.                     cond: {
320.                       $gte: [
321.                         {
322.                           $dateDiff: {
323.                             startDate: "$$this.firstLogin",
324.                             endDate: "$$this.lastLogin",
325.                             unit: "day"
326.                           }
327.                         },
328.                         7
329.                       ]
330.                     }
331.                   }
332.                 }
333.               },
334.               "$userCount"
335.             ]
336.           },
337.           day30Retention: {
338.             $divide: [
339.               {
340.                 $size: {
341.                   $filter: {
342.                     input: "$users",
343.                     cond: {
344.                       $gte: [
345.                         {
346.                           $dateDiff: {
347.                             startDate: "$$this.firstLogin",
348.                             endDate: "$$this.lastLogin",
349.                             unit: "day"
350.                           }
351.                         },
352.                         30
353.                       ]
354.                     }
355.                   }
356.                 }
357.               },
358.               "$userCount"
359.             ]
360.           }
361.         }
362.       },
363.       {
364.         $sort: { cohort: 1 }
365.       }
366.     ];
367.    
368.     const userRetention = await database.collection("user_events").aggregate(userRetentionPipeline).toArray();
369.    
370.     // 导出用户留存数据
371.     fs.writeFileSync('user_retention_analysis.json', JSON.stringify(userRetention, null, 2));
372.     console.log(`已导出 ${userRetention.length} 个用户群组的留存分析数据`);
373.    
374.     // 生成分析报告
375.     const report = `# 用户行为分析报告
377. ## 分析期间
378. 最近30天（${new Date(new Date() - 30 * 24 * 60 * 60 * 1000).toISOString().split('T')[0]} 至 ${new Date().toISOString().split('T')[0]}）
380. ## 用户活动分析
381. - 分析用户数: ${userActivity.length}
382. - 平均每个用户会话数: ${(userActivity.reduce((sum, user) => sum + user.sessionCount, 0) / userActivity.length).toFixed(2)}
383. - 最活跃用户: ${userActivity[0].userId} (${userActivity[0].sessionCount} 次会话)
385. ## 用户路径分析
386. - 分析用户数: ${userPaths.length}
387. - 平均每个用户页面浏览数: ${(userPaths.reduce((sum, user) => sum + user.sessionCount, 0) / userPaths.length).toFixed(2)}
388. - 平均每个用户停留时间: ${(userPaths.reduce((sum, user) => sum + user.timeSpent, 0) / userPaths.length / 1000 / 60).toFixed(2)} 分钟
390. ## 转化漏斗分析
391. `;
392.    
393.     if (conversionFunnel.length > 0) {
394.       const funnel = conversionFunnel[0];
395.       report += `
396. - 总用户数: ${funnel.totalUsers}
397. - 首页访问数: ${funnel.homepageVisits} (${(funnel.homepageConversionRate * 100).toFixed(2)}%)
398. - 产品浏览数: ${funnel.productViews} (${(funnel.productViewRate * 100).toFixed(2)}%)
399. - 加入购物车数: ${funnel.cartAdditions} (${(funnel.cartAdditionRate * 100).toFixed(2)}%)
400. - 结账开始数: ${funnel.checkouts} (${(funnel.checkoutRate * 100).toFixed(2)}%)
401. - 完成购买数: ${funnel.purchases} (${(funnel.purchaseRate * 100).toFixed(2)}%)
402. - 整体转化率: ${(funnel.overallConversionRate * 100).toFixed(2)}%
403. `;
404.     }
405.    
406.     report += "\n## 用户留存分析\n";
407.    
408.     if (userRetention.length > 0) {
409.       report += "| 群组 | 用户数 | 次日留存率 | 7日留存率 | 30日留存率 |\n";
410.       report += "|------|--------|------------|-----------|------------|\n";
411.       
412.       userRetention.forEach(cohort => {
413.         report += `| ${cohort.cohort} | ${cohort.userCount} | ${(cohort.day1Retention * 100).toFixed(2)}% | ${(cohort.day7Retention * 100).toFixed(2)}% | ${(cohort.day30Retention * 100).toFixed(2)}% |\n`;
414.       });
415.     }
416.    
417.     // 保存报告
418.     fs.writeFileSync('user_behavior_analysis_report.md', report);
419.    
420.     console.log("用户行为分析完成，已生成以下文件:");
421.     console.log("- user_activity_analysis.json: 用户活动分析数据");
422.     console.log("- user_path_analysis.json: 用户路径分析数据");
423.     console.log("- conversion_funnel_analysis.json: 转化漏斗分析数据");
424.     console.log("- user_retention_analysis.json: 用户留存分析数据");
425.     console.log("- user_behavior_analysis_report.md: 用户行为分析报告");
426.    
427.   } finally {
428.     await client.close();
429.   }
430. }
432. userBehaviorAnalysis().catch(console.error);

复制代码

7. 总结

本文详细介绍了从MongoDB数据库导出JSON数据的各种方法、工具和最佳实践。我们讨论了：

1. 多种导出方法：使用mongoexport命令行工具使用MongoDB Compass图形界面使用编程语言驱动程序（Node.js、Python、Java）使用聚合管道进行数据转换
2. 使用mongoexport命令行工具
3. 使用MongoDB Compass图形界面
4. 使用编程语言驱动程序（Node.js、Python、Java）
5. 使用聚合管道进行数据转换
6. 最佳实践：性能优化技术，包括批量处理和索引优化数据安全措施，包括敏感数据处理和访问控制大数据集处理策略，包括分片集合导出和增量导出
7. 性能优化技术，包括批量处理和索引优化
8. 数据安全措施，包括敏感数据处理和访问控制
9. 大数据集处理策略，包括分片集合导出和增量导出
10. 常见问题和解决方案：处理特殊数据类型（ObjectId、Date等）处理大数据量和内存限制处理连接和认证问题
11. 处理特殊数据类型（ObjectId、Date等）
12. 处理大数据量和内存限制
13. 处理连接和认证问题
14. 实际应用场景：数据备份与恢复数据迁移（MongoDB到其他数据库，其他数据库到MongoDB）数据分析和报告生成
15. 数据备份与恢复
16. 数据迁移（MongoDB到其他数据库，其他数据库到MongoDB）
17. 数据分析和报告生成

多种导出方法：

• 使用mongoexport命令行工具
• 使用MongoDB Compass图形界面
• 使用编程语言驱动程序（Node.js、Python、Java）
• 使用聚合管道进行数据转换

最佳实践：

• 性能优化技术，包括批量处理和索引优化
• 数据安全措施，包括敏感数据处理和访问控制
• 大数据集处理策略，包括分片集合导出和增量导出

常见问题和解决方案：

• 处理特殊数据类型（ObjectId、Date等）
• 处理大数据量和内存限制
• 处理连接和认证问题

实际应用场景：

• 数据备份与恢复
• 数据迁移（MongoDB到其他数据库，其他数据库到MongoDB）
• 数据分析和报告生成

通过本文提供的详细指南和代码示例，开发者可以根据自己的需求选择最适合的方法来导出MongoDB数据为JSON格式，并确保导出过程高效、安全且可靠。

无论您是需要进行数据备份、系统迁移、数据分析还是其他任何需要导出MongoDB数据的场景，本文提供的知识都将帮助您轻松实现数据导出与交换。