如何将FTP服务器每日CSV数据接入MongoDB Atlas并自动同步集合？

核心思路 #

MongoDB Atlas本身不支持直接连接FTP服务器，需要搭建一个中间服务来完成：FTP文件拉取 → 关系型CSV转MongoDB文档 → 写入/更新Atlas的完整流程。

步骤1：搭建中间同步服务（以Python为例） #

Python生态有成熟的库处理FTP和MongoDB，适合快速实现：

依赖安装 #

pip install pymongo python-dotenv pandas

核心代码逻辑 #

1. FTP拉取CSV文件

import ftplib
import os
from dotenv import load_dotenv

load_dotenv()

# FTP配置
FTP_HOST = os.getenv("FTP_HOST")
FTP_USER = os.getenv("FTP_USER")
FTP_PASS = os.getenv("FTP_PASS")
CSV_FILE_NAME = "daily_data.csv"
LOCAL_CSV_PATH = "./temp_data.csv"

# 连接FTP并下载文件
with ftplib.FTP(FTP_HOST) as ftp:
    ftp.login(FTP_USER, FTP_PASS)
    with open(LOCAL_CSV_PATH, "wb") as f:
        ftp.retrbinary(f"RETR {CSV_FILE_NAME}", f.write)

2. 处理关系型CSV转MongoDB文档
   如果CSV是关系型结构（比如含主外键的单表或拆分的多表），需要适配MongoDB的文档模型：

• 若为单表含主外键：将关联数据嵌套为数组（比如把订单明细嵌套进对应订单文档的items字段）
• 若为多张独立CSV：可分别创建集合，用唯一字段做引用，但更推荐嵌套结构以发挥MongoDB的文档优势

示例：转换含订单与明细的CSV（以order_id为关联键）

import pandas as pd

df = pd.read_csv(LOCAL_CSV_PATH)
# 按order_id分组，将明细嵌套为数组
grouped_data = df.groupby("order_id").apply(lambda x: x.to_dict("records")).reset_index(name="items")
# 转换为MongoDB可接受的文档格式
docs = grouped_data.to_dict("records")

3. 连接Atlas并写入/更新数据

from pymongo import MongoClient
from pymongo.errors import PyMongoError

# Atlas连接配置
ATLAS_URI = os.getenv("ATLAS_URI")
DB_NAME = "your_database"
COLLECTION_NAME = "orders"

client = MongoClient(ATLAS_URI)
db = client[DB_NAME]
collection = db[COLLECTION_NAME]

# 批量写入/更新（以order_id为唯一键）
for doc in docs:
    try:
        # 存在则更新，不存在则插入
        collection.update_one(
            {"order_id": doc["order_id"]},
            {"$set": doc},
            upsert=True
        )
    except PyMongoError as e:
        print(f"处理文档order_id={doc['order_id']}失败: {str(e)}")

client.close()

步骤2：实现自动触发 #

有两种常见方式实现每日自动同步：

• 定时任务：

  • Linux/macOS：用cron，编辑定时任务：

    crontab -e
    # 添加每日凌晨2点执行脚本
    0 2 * * * /usr/bin/python3 /path/to/your/sync_script.py >> /path/to/logs/sync.log 2>&1
    

  • Windows：用「任务计划程序」创建每日触发的任务，执行Python脚本。

• 监听FTP文件变化：
  若需要在FTP服务器上传文件后立即触发同步，可使用watchdog库监听FTP服务器的文件目录（需FTP支持被动模式且可访问目录），或配置FTP服务器在文件上传后触发脚本。

步骤3：关键注意事项 #

• Atlas权限配置：将中间服务的公网IP添加到Atlas的IP白名单中，并创建具有readWrite权限的数据库用户。
• 数据类型处理：CSV默认是字符串类型，需手动转换日期、数字等类型（比如用pandas的to_datetime、astype方法）。
• 容错与日志：添加异常捕获，记录同步日志，方便排查失败原因；可实现失败重试机制。
• 关系型结构适配：如果必须保留关系型引用，可在MongoDB中用DBRef，但更推荐根据业务场景设计嵌套文档模型，提升查询效率。

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