如何在Python中高效将MongoDB集合导入DuckDB以降低内存消耗？

1. 直接对接MongoDB与DuckDB（最优方案） #

DuckDB官方提供MongoDB扩展，可直接从MongoDB集合读取数据创建表，完全跳过中间JSONL文件，从根源避免内存过载问题。

代码示例：

import duckdb

# 初始化DuckDB连接
con = duckdb.connect()

# 安装并加载MongoDB扩展（首次运行需安装，后续可跳过INSTALL步骤）
con.sql("INSTALL mongodb;")
con.sql("LOAD mongodb;")

# 直接读取MongoDB数据创建DuckDB表
# 替换为你的MongoDB连接串、数据库名和集合名
con.sql("""
CREATE OR REPLACE TABLE mongo_table AS
SELECT * FROM mongodb_scan(
    'mongodb://localhost:27017',
    'your_database_name',
    'your_collection_name'
)
""")

# 验证数据导入结果
print(con.sql("SELECT COUNT(*) FROM mongo_table").fetchone())

2. 分批次读取MongoDB并写入DuckDB #

如果无法使用扩展，可分批次从MongoDB拉取数据，逐批插入DuckDB，避免一次性加载全量数据到内存。

代码示例：

import duckdb
from pymongo import MongoClient

# 初始化连接
mongo_client = MongoClient("mongodb://localhost:27017")
mongo_db = mongo_client["your_database_name"]
mongo_coll = mongo_db["your_collection_name"]
con = duckdb.connect()

# 先创建表结构（用单条数据自动推断）
sample = list(mongo_coll.find().limit(1))
con.sql("CREATE OR REPLACE TABLE mongo_table AS SELECT * FROM?", parameters=(sample,))

# 分批次插入数据，批次大小可根据内存调整
batch_size = 1000
skip = 0
while True:
    batch = list(mongo_coll.find().skip(skip).limit(batch_size))
    if not batch:
        break
    con.sql("INSERT INTO mongo_table SELECT * FROM?", parameters=(batch,))
    skip += batch_size
    print(f"已插入 {skip} 条数据")

# 关闭连接
mongo_client.close()
con.close()

3. 流式生成JSONL+分块读取 #

若必须保留JSONL中间文件，优化写入和读取逻辑：

• 写入时逐行生成JSONL，避免一次性加载全量数据到内存
• 读取时通过DuckDB参数控制分块加载，降低内存占用

流式写入JSONL #

import json
from pymongo import MongoClient

mongo_client = MongoClient("mongodb://localhost:27017")
mongo_db = mongo_client["your_database_name"]
mongo_coll = mongo_db["your_collection_name"]

# 逐行写入，不加载全量数据到内存
with open("mongo_json.jsonl", "w") as file:
    for doc in mongo_coll.find():
        doc["_id"] = str(doc["_id"])  # 转换MongoDB特殊类型为字符串
        file.write(json.dumps(doc) + "\n")

mongo_client.close()

DuckDB分块读取JSONL #

import duckdb

con = duckdb.connect()

# 用ROWS_PER_CHUNK控制每次加载的行数，减少内存占用
con.sql("""
CREATE OR REPLACE TABLE mongo_table AS
SELECT * FROM read_json_auto(
    'mongo_json.jsonl',
    IGNORE_ERRORS=true,
    ROWS_PER_CHUNK=10000  # 可根据内存调整chunk大小
)
""")

4. 拆分JSONL为多个小文件 #

使用MongoDB官方mongoexport命令行工具，直接导出为拆分的多份JSONL文件，DuckDB支持读取通配符匹配的多个文件，避免单个大文件的内存压力。

导出拆分文件（命令行） #

mongoexport --uri "mongodb://localhost:27017/your_database_name" \
            --collection your_collection_name \
            --out part_ \
            --splitSize 1GB  # 每个文件最大1GB，可按需调整

DuckDB读取多份文件 #

import duckdb

con = duckdb.connect()
con.sql("""
CREATE OR REPLACE TABLE mongo_table AS
SELECT * FROM read_json_auto('part_*.jsonl', IGNORE_ERRORS=true)
""")

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