从MongoDB到PostgreSQL：MosQL实时数据同步实战指南

引言：当NoSQL遇上SQL的痛点与解决方案

你是否正面临这样的困境：MongoDB的灵活schema加速了开发迭代，却让数据分析团队在复杂查询和多表关联面前束手无策？作为开发者，你是否既想要文档数据库的schema灵活性，又需要关系型数据库的强大查询能力？MosQL（MongoDB to SQL） 正是为解决这一矛盾而生的开源工具，它能将MongoDB数据实时同步到PostgreSQL，让你同时享受两种数据库的优势。

本文将提供一份全面的MosQL实战指南，包含从环境搭建到高级配置的完整流程。通过阅读本文，你将掌握：

• MosQL的核心工作原理与架构设计
• 从零开始的安装与配置步骤
• 复杂数据类型的映射与转换技巧
• 高可用同步方案的实现策略
• 常见问题的诊断与优化方法

MosQL核心架构与工作原理

数据同步流程图

flowchart TD
    subgraph MongoDB集群
        A[Primary节点] -->|写入操作| B[Oplog]
        C[Secondary节点] -->|复制| B
    end
    D[MosQL Tailer] -->|监听| B
    D -->|解析操作| E[数据转换器]
    E -->|类型映射| F[PostgreSQL适配器]
    subgraph PostgreSQL
        G[目标表] <--|UPSERT| F
        H[同步状态表] <--|记录位置| D
    end
    I[初始数据导入] -->|批量写入| G

核心组件解析

MosQL主要由四个模块构成：

1. Tailer模块：通过监听MongoDB的Oplog（操作日志）捕获实时变更，支持断点续传
2. Schema模块：定义MongoDB文档到PostgreSQL表的映射规则，处理数据类型转换
3. Streamer模块：负责初始数据全量导入和增量同步的协调工作
4. SQL适配器：处理与PostgreSQL的交互，包括表创建、数据写入和冲突解决

环境准备与安装指南

系统要求

软件	最低版本	推荐版本	作用
Ruby	2.3.0	2.7.0+	运行MosQL核心程序
PostgreSQL	9.3	13.0+	目标数据库，支持JSON/JSONB类型
MongoDB	3.0	4.4+	源数据库，需配置为副本集
libpq-dev	-	最新版	PostgreSQL客户端开发库
zlib1g-dev	-	最新版	数据压缩支持

安装步骤

1. 通过源码安装（推荐）

# 克隆仓库
git clone https://gitcode.com/gh_mirrors/mo/mosql
cd mosql

# 安装依赖
bundle install

# 构建并安装gem
gem build mosql.gemspec
gem install mosql-*.gem

2. 通过RubyGems安装

gem install mosql

注意：MosQL目前不再积极维护，建议从源码安装以获取最新修复

配置文件详解：Collection Map

Collection Map是MosQL的核心配置文件，采用YAML格式定义MongoDB到PostgreSQL的映射关系。

基础配置结构

# collections.yml
mongodb_database_name:
  mongodb_collection_name:
    :meta:
      :table: postgresql_table_name  # 目标表名
      :extra_props: true             # 是否保留未映射字段
      :composite_key: [col1, col2]   # 复合主键（可选）
    :columns:
      - _id: TEXT                    # 简单映射（字段名和类型）
      - user_name:
          :source: name              # MongoDB源字段
          :type: VARCHAR(100)        # PostgreSQL数据类型
      - user_age:
          :source: profile.age       # 嵌套字段（点表示法）
          :type: INTEGER
      - tags: TEXT ARRAY             # 数组类型

数据类型映射表

MongoDB类型	PostgreSQL类型	转换规则	示例
ObjectId	TEXT	转换为字符串	507f1f77bcf86cd799439011 → "507f1f77bcf86cd799439011"
String	TEXT/VARCHAR	直接映射	"hello" → "hello"
Integer	INTEGER/BIGINT	直接映射	42 → 42
Float	DOUBLE PRECISION	直接映射	3.14 → 3.14
Boolean	BOOLEAN	直接映射	true → true
Array	TEXT ARRAY	元素类型转换后保持数组结构	[1, 2, 3] → {1,2,3}
Embedded Document	JSONB	序列化为JSON字符串	{a: 1, b: 2} → '{"a":1,"b":2}'
Date	TIMESTAMP	保留时间精度	ISODate("2023-01-01") → 2023-01-01 00:00:00
Binary	BYTEA	保持二进制数据	BinData(0, "AA==") → \x4141

高级配置示例

1. 处理嵌套文档

products:
  items:
    :meta:
      :table: products
    :columns:
      - _id: TEXT
      - product_name:
          :source: name
          :type: VARCHAR(200)
      - price:
          :source: pricing.current
          :type: DECIMAL(10,2)
      - category:
          :source: attributes.category
          :type: TEXT

2. 数组与特殊类型处理

users:
  profiles:
    :meta:
      :table: user_profiles
      :extra_props: JSONB  # 使用PostgreSQL JSONB类型存储额外字段
    :columns:
      - _id: TEXT
      - tags: TEXT ARRAY  # 字符串数组
      - permissions: INTEGER ARRAY  # 整数数组
      - last_login: TIMESTAMP
      - avatar: BYTEA  # 二进制数据

基本操作指南

命令行参数概览

mosql [选项]
  -c, --collections FILE    指定配置文件（默认：collections.yml）
  --sql URI                 PostgreSQL连接URI（默认：postgres:///）
  --mongo URI               MongoDB连接URI（默认：mongodb://localhost）
  --skip-tail               仅执行初始导入，不启动实时同步
  --skip-import             跳过初始导入，直接启动实时同步
  --reimport                强制重新导入数据（删除现有表）
  --verbose                 显示详细日志
  --unsafe                  忽略插入错误继续执行

连接字符串格式

数据库	连接字符串示例
PostgreSQL	postgres://user:password@host:port/dbname
MongoDB	mongodb://user:password@host:port/?readPreference=secondary

典型工作流程

1. 完整同步（初始导入+实时同步）

mosql --collections my_collections.yml \
      --sql postgres://postgres@localhost/mydb \
      --mongo mongodb://mongodb:27017/?readPreference=secondary

2. 仅执行初始数据导入

mosql --collections my_collections.yml \
      --sql postgres://postgres@localhost/mydb \
      --mongo mongodb://mongodb:27017/ \
      --skip-tail

3. 从指定时间点开始同步

mosql --collections my_collections.yml \
      --sql postgres://postgres@localhost/mydb \
      --mongo mongodb://mongodb:27017/ \
      --tail-from 1620000000  # Unix时间戳

数据类型转换深度解析

复杂类型转换规则

MosQL对MongoDB复杂类型有特殊处理逻辑，确保数据在PostgreSQL中保持可用性：

1. 嵌套文档转换

MongoDB嵌套文档默认会被序列化为JSON字符串存储：

// MongoDB文档
{
  "_id": ObjectId("5f8d0d55b6b5a432a4c3a1b2c"),
  "user": {
    "name": "John Doe",
    "address": {
      "city": "New York",
      "zip": "10001"
    }
  }
}

// 转换后PostgreSQL记录（无显式映射时）
{
  "_id": "5f8d0d55b6b5a432a4c3a1b2c",
  "_extra_props": {
    "user": {
      "name": "John Doe",
      "address": {
        "city": "New York",
        "zip": "10001"
      }
    }
  }
}

2. 数组类型处理

数组根据配置有两种处理方式：

# 1. 映射为PostgreSQL数组类型
- tags: TEXT ARRAY

# 2. 序列化为JSON字符串
- tags: TEXT

3. 特殊BSON类型转换

BSON类型	转换方式	PostgreSQL类型建议
ObjectId	转为字符串	TEXT
Binary	保留二进制或Base64编码	BYTEA或TEXT
DBRef	仅保留ObjectId部分	TEXT
Timestamp	转为ISO格式时间戳	TIMESTAMP
Decimal128	转为字符串	DECIMAL或TEXT

数据转换示例

mindmap
  root((数据转换流程))
    MongoDB输入
      ObjectId
      嵌套文档
      数组
      二进制数据
    转换处理
      类型检测
      格式转换
      错误处理
    PostgreSQL输出
      TEXT
      JSONB
      ARRAY
      BYTEA

高级功能与最佳实践

处理Schema不匹配

当MongoDB文档结构与配置的映射规则不匹配时，MosQL提供多种处理策略：

1. 忽略错误继续执行：使用--unsafe参数
2. 记录错误日志并跳过：默认行为
3. 自定义错误处理：通过修改源码实现（lib/mosql/streamer.rb中的unsafe_handle_exceptions方法）

性能优化策略

1. 批量处理配置

# lib/mosql/streamer.rb 中的常量定义
BATCH = 1000  # 批量插入大小，可根据服务器性能调整

2. 连接池与资源控制

# 调整PostgreSQL连接数
mosql --sql "postgres://user@host/db?pool=10"  # 设置连接池大小

3. 同步策略选择

场景	推荐策略	优势
大数据量初始导入	--skip-tail + 后续同步	避免同步延迟累积
生产环境实时同步	从Secondary节点读取	减轻主库压力
低延迟要求	减小批量大小	提高同步频率

高可用部署方案

sequenceDiagram
    participant M1 as MongoDB Primary
    participant M2 as MongoDB Secondary
    participant M3 as MongoDB Arbiter
    participant MS as MosQL实例
    participant P1 as PostgreSQL Primary
    participant P2 as PostgreSQL Standby
    
    M1->>M2: 复制数据
    MS->>M2: 读取Oplog（读偏好设置）
    MS->>P1: 写入数据
    P1->>P2: 流复制

推荐配置：

• MongoDB使用副本集，MosQL连接到Secondary节点
• PostgreSQL配置主从复制，MosQL写入主库
• 监控同步延迟，设置告警阈值（如延迟>30秒）

常见问题解决方案

1. MongoDB认证配置

# 使用认证连接MongoDB
mosql --mongo "mongodb://user:password@host:port/admin?authSource=admin"

2. 处理分片集群

MosQL不直接支持分片集群，需为每个分片单独部署实例：

# 分片1同步
mosql --mongo "mongodb://shard1-host:27017/" --sql "postgres:///shard1_db"

# 分片2同步
mosql --mongo "mongodb://shard2-host:27017/" --sql "postgres:///shard2_db"

3. 数据一致性保障

# 定期验证数据一致性（外部脚本）
compare_mongo_postgres.sh  # 对比两边记录数和关键字段

实战案例：电子商务数据同步

场景描述

某电商平台使用MongoDB存储订单数据，需要同步到PostgreSQL进行销售分析。订单数据包含嵌套的商品列表、用户信息和支付详情。

配置文件示例

# order_sync.yml
ecommerce:
  orders:
    :meta:
      :table: orders
      :extra_props: JSONB
    :columns:
      - _id: TEXT
      - order_number:
          :source: orderNo
          :type: VARCHAR(50)
      - user_id:
          :source: user._id
          :type: TEXT
      - total_amount:
          :source: payment.amount
          :type: DECIMAL(12,2)
      - status:
          :source: status
          :type: TEXT
      - created_at:
          :source: createdAt
          :type: TIMESTAMP
      - items:
          :source: products
          :type: JSONB  # 将商品列表存储为JSONB

执行同步命令

mosql --collections order_sync.yml \
      --sql "postgres://analytics@pg-host/ecom_analytics" \
      --mongo "mongodb://mongo-host/ecommerce?readPreference=secondary" \
      --verbose

数据分析查询示例

-- 分析各商品类别的销售情况
SELECT 
  item->>'category' as category,
  COUNT(*) as order_count,
  SUM((item->>'price')::decimal * (item->>'quantity')::int) as total_sales
FROM 
  orders,
  jsonb_array_elements(items) as item
WHERE 
  created_at >= '2023-01-01'
GROUP BY 
  category
ORDER BY 
  total_sales DESC;

故障排除与常见问题

连接问题

错误信息	可能原因	解决方案
Mongo::ConnectionFailure	MongoDB连接失败	检查URI格式和服务状态
Sequel::DatabaseConnectionError	PostgreSQL连接失败	验证凭据和网络可达性
Authentication failed	认证失败	检查用户名密码和authSource

数据同步问题

1. 同步延迟不断增加

可能原因：

• 批量大小设置过大
• 目标数据库写入性能不足
• MongoDB查询性能问题

解决方案：

# 减小批量大小（需要修改源码后重新构建）
# 在lib/mosql/streamer.rb中修改BATCH常量
BATCH = 500  # 减小为原来的一半

2. 数据类型转换错误

示例错误：PG::InvalidTextRepresentation: ERROR: invalid input syntax for integer

解决方案：

• 检查MongoDB文档中对应字段的实际类型
• 修改配置文件，使用更合适的数据类型（如从INTEGER改为TEXT）
• 使用_extra_props捕获非标准字段

监控与告警

推荐监控指标：

1. 同步延迟（MosQL记录的最后Oplog时间与当前时间差）
2. 错误日志出现频率
3. 数据库连接数和资源使用率

总结与展望

MosQL作为MongoDB到PostgreSQL的实时同步工具，填补了文档数据库与关系型数据库之间的数据流动鸿沟。尽管该项目目前不再积极维护，但其核心思想和实现方式仍具有重要参考价值。

项目优势

• 轻量级设计，易于部署和配置
• 灵活的数据映射规则，支持复杂数据类型
• 实时同步与初始导入一体化解决方案

潜在改进方向

1. 支持MongoDB分片集群：自动发现并同步所有分片
2. 更丰富的数据转换函数：自定义转换逻辑
3. Web管理界面：可视化配置和监控
4. 增量初始导入：支持基于时间范围的部分导入

相关替代工具

工具	语言	特点
Debezium	Java	基于CDC的通用变更捕获平台
Mongo-Connector	Python	MongoDB官方同步工具
pg_chameleon	Python	专注于MySQL到PostgreSQL同步

通过本文的指南，你应该能够顺利部署和使用MosQL，实现MongoDB与PostgreSQL之间的无缝数据同步。无论是构建实时数据分析平台，还是实现多数据库架构，MosQL都能为你提供灵活可靠的数据流动能力。