AWS EMR中PySpark/Python读取MongoDB副本集异常求助

一、PySpark读取返回空DataFrame问题 #

1. 修正URI配置错误 #

• 检查URI中的用户名拼写（你的代码里写的是usename，缺少字母r），确认密码、主机端口、副本集名称ABCD与MongoDB配置完全一致。
• 确认authSource参数：若数据库用户是在目标库而非admin创建，需将authSource=admin改为对应数据库名。

2. 规范SparkSession初始化与读取配置 #

• Spark 2.x+无需手动创建SparkContext，直接通过SparkSession.builder构建，避免上下文冲突：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder.appName("MongoDbToS3")\
      .config("spark.mongodb.input.uri", "mongodb://username:password@host1:port1,host2:port2,host3:port3/db.collection/?replicaSet=ABCD&authSource=admin")\
      .config("spark.mongodb.input.database", "db")\
      .config("spark.mongodb.input.collection", "collection")\
      .getOrCreate()
  data = spark.read.format("com.mongodb.spark.sql.DefaultSource").load()
  data.show()
  

• 显式指定spark.mongodb.input.database和spark.mongodb.input.collection，避免URI解析异常。

3. 验证权限与副本集状态 #

• 在MongoDB shell中用指定用户登录，执行db.collection.find().limit(1)，确认用户有目标集合的读权限。
• 执行rs.status()检查副本集节点状态，确保所有节点处于PRIMARY或SECONDARY状态，无不可用节点。

4. 分析Spark日志细节 #

• 查看Spark executor日志，搜索MongoDB相关条目，确认是否存在查询过滤条件误配置，或连接到错误的数据库/集合。

二、pymongo读取卡顿/超时问题 #

1. 优化副本集连接配置 #

• 连接字符串需包含所有副本集节点，而非单个host，补充关键超时参数：

  import pymongo
  
  client = pymongo.MongoClient(
      "mongodb://username:password@host1:port1,host2:port2,host3:port3/?replicaSet=ABCD&authSource=admin",
      socketTimeoutMS=3600000,
      connectTimeoutMS=3600000,
      serverSelectionTimeoutMS=3600000,
      maxPoolSize=10
  )
  

2. 针对大文档的处理策略 #

• 只读取所需字段，减少数据传输量：

  # 示例：仅读取_id和需要的业务字段
  results = quoteinfo__collection.find({}, {"_id": 1, "target_field": 1}).batch_size(100)
  

• 禁用游标超时（需手动关闭游标避免资源泄漏）：

  results = quoteinfo__collection.find({}, no_cursor_timeout=True).batch_size(100)
  # 处理完成后务必关闭
  results.close()
  

3. 网络与MongoDB端排查 #

• 测试EMR与MongoDB节点间的网络带宽：用scp传输大文件，确认带宽是否满足大文档传输需求。
• 开启MongoDB慢查询分析：

  // 开启profiling（级别2记录所有操作）
  db.setProfilingLevel(2)
  // 查看最近的读操作日志
  db.system.profile.find({op: "query", ns: "database_name.collection_name"}).sort({ts: -1}).limit(10)
  

• 检查MongoDB的maxMessageSizeBytes配置（默认48MB），若文档超过该值会导致传输失败，需修改配置后重启MongoDB。

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