如何使用Azure Data Factory将单个MongoDB集合的数据迁移至PostgreSQL多表

方法1：基于Azure Data Factory（ADF）数据流实现 #

• 源数据配置：创建MongoDB源数据集，指向目标集合，支持全量或增量数据读取。
• 数据拆分：在数据流中添加拆分转换，依据业务规则将单集合数据拆分为多个数据流分支。比如按文档内的entity_type字段值拆分，或从嵌套文档中提取不同实体数据（如从包含用户+订单的文档中拆分出用户流和订单流）。
• 字段映射与转换：为每个分支添加映射转换，调整字段名称、数据类型，匹配PostgreSQL目标表的结构（例如将MongoDB的_id映射为PostgreSQL的user_id）。
• 多表写入：为每个分支创建对应的PostgreSQL目标数据集，指定不同目标表，设置写入模式（追加/覆盖等），完成数据分流写入。

方法2：自定义脚本实现（以Python为例） #

• 读取MongoDB数据：用MongoDB驱动批量读取目标集合数据

  from pymongo import MongoClient
  client = MongoClient("mongodb://your-host:27017/")
  db = client["target-db"]
  collection = db["source-collection"]
  docs = list(collection.find({}))
  

• 数据拆分与预处理：遍历文档，按业务逻辑拆分出不同表所需数据，同步完成类型适配（如MongoDB ObjectId转字符串、日期格式对齐PostgreSQL）。
• 批量写入PostgreSQL：用PostgreSQL驱动连接数据库，通过批量插入语句将拆分后的数据分别写入对应表

  import psycopg2
  conn = psycopg2.connect("dbname=target-db user=your-user password=your-pwd host=your-host")
  cur = conn.cursor()
  
  # 写入用户表示例
  user_records = [(doc["_id"], doc["username"], doc["email"]) for doc in docs]
  cur.executemany("INSERT INTO users (id, username, email) VALUES (%s, %s, %s)", user_records)
  
  # 写入关联订单表示例
  order_records = [(doc["_id"], item["order_id"], item["amount"]) for doc in docs for item in doc["orders"]]
  cur.executemany("INSERT INTO orders (user_id, order_id, amount) VALUES (%s, %s, %s)", order_records)
  
  conn.commit()
  cur.close()
  conn.close()
  

方法3：使用传统ETL工具（如Talend、Informatica） #

• 配置MongoDB源连接与多个PostgreSQL目标连接；
• 拖拽数据提取组件读取集合数据，添加路由组件按规则拆分数据流，再通过多个数据加载组件分别写入对应的PostgreSQL表。

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