MongoDB With Spark遇到的2个错误，不能初始化和sample重复的key

１.$sample stage could not find a non-duplicate document while using a random cursor

　　这个问题比较难解决，因为我用mongodb spark connector没用到sample，但是在生成RDD的过程中会进行sample操作，所以没法避免，出现这个问题的原因也不可控，在jira上有这个问题，但并没有一个合理的解决方案，stackoverflow上也没有解决办法，就我个人而言，出现这个问题有几个特征：

　　a) 出现在sharding 集群中，因为我单机不用spark，也不会出现这种问题，而在搜索过程中发现遇到这问题的人非常少，而且很多都不可再现，而spark连sharding集群会一直出这个问题。

　　b) 似乎和shardsvr有直接联系，出现这问题的机器一直是其中一台shardsvr，而之前同样的程序在集群中是跑通过的，所以怀疑是上次异常断电导致这台shardsvr有什么设置出现了问题。

　　解决方案目前可以使用sparkSession.config设置这项参数，也可以直接在readConfig里设置这项参数，如sparkSession进行如下设置：

	lazy val spark = SparkSession.builder()
		.master("spark://192.168.12.161:7077")
		.config(new SparkConf().setJars(Array("hdfs://192.168.12.161:9000/mongolib/mongo-spark-connector_2.11-2.0.0.jar",
				"hdfs://192.168.12.161:9000/mongolib/bson-3.4.2.jar",
				"hdfs://192.168.12.161:9000/mongolib/mongo-java-driver-3.4.2.jar",
				"hdfs://192.168.12.161:9000/mongolib/mongodb-driver-3.4.2.jar",
				"hdfs://192.168.12.161:9000/mongolib/mongodb-driver-core-3.4.2.jar",
				"hdfs://192.168.12.161:9000/mongolib/commons-io-2.5.jar",
				"hdfs://192.168.12.161:9000/mongolib/config-1.2.1.jar",
				"hdfs://192.168.12.161:9000/ScheduleImport.jar")))  	  
		.config("spark.cores.max", 80)		
		.config("spark.executor.cores", 16)
		.config("spark.executor.memory", "32g")
		.config("spark.mongodb.input.uri", s"mongodb://${user}:${passwd}@192.168.12.161:27017/wenshu.origin2?authSource=${authDB}")
		.config("spark.mongodb.output.uri", s"mongodb://${user}:${passwd}@192.168.12.160:27017/wenshu.backup?authSource=${authDB}")
		.config("spark.mongodb.input.partitionerOptions.samplesPerPartition", 1)
		.getOrCreate()

　　　这个参数在Connector中的作用如下：

 override def partitions(connector: MongoConnector, readConfig: ReadConfig, pipeline: Array[BsonDocument]): Array[MongoPartition] = {
    Try(PartitionerHelper.collStats(connector, readConfig)) match {
      case Success(results) =>
        val matchQuery = PartitionerHelper.matchQuery(pipeline)
        val partitionerOptions = readConfig.partitionerOptions.map(kv => (kv._1.toLowerCase, kv._2))
        val partitionKey = partitionerOptions.getOrElse(partitionKeyProperty, DefaultPartitionKey)
        val partitionSizeInBytes = partitionerOptions.getOrElse(partitionSizeMBProperty, DefaultPartitionSizeMB).toInt * 1024 * 1024
        val samplesPerPartition = partitionerOptions.getOrElse(samplesPerPartitionProperty, DefaultSamplesPerPartition).toInt

        val count = if (matchQuery.isEmpty) {
          results.getNumber("count").longValue()
        } else {
          connector.withCollectionDo(readConfig, { coll: MongoCollection[BsonDocument] => coll.count(matchQuery) })
        }
        val avgObjSizeInBytes = results.get("avgObjSize", new BsonInt64(0)).asNumber().longValue()
        val numDocumentsPerPartition: Int = math.floor(partitionSizeInBytes.toFloat / avgObjSizeInBytes).toInt
        val numberOfSamples = math.floor(samplesPerPartition * count / numDocumentsPerPartition.toFloat).toInt

        if (numDocumentsPerPartition >= count) {
          MongoSinglePartitioner.partitions(connector, readConfig, pipeline)
        } else {
          val samples = connector.withCollectionDo(readConfig, {
            coll: MongoCollection[BsonDocument] =>
              coll.aggregate(List(
                Aggregates.`match`(matchQuery),
                Aggregates.sample(numberOfSamples),
                Aggregates.project(Projections.include(partitionKey)),
                Aggregates.sort(Sorts.ascending(partitionKey))
              ).asJava).allowDiskUse(true).into(new util.ArrayList[BsonDocument]()).asScala
          })
          def collectSplit(i: Int): Boolean = (i % samplesPerPartition == 0) || !matchQuery.isEmpty && i == count - 1
          val rightHandBoundaries = samples.zipWithIndex.collect {
            case (field, i) if collectSplit(i) => field.get(partitionKey)
          }
          val addMinMax = matchQuery.isEmpty
          val partitions = PartitionerHelper.createPartitions(partitionKey, rightHandBoundaries, PartitionerHelper.locations(connector), addMinMax)
          if (!addMinMax) PartitionerHelper.setLastBoundaryToLessThanOrEqualTo(partitionKey, partitions)
          partitions
        }

      case Failure(ex: MongoCommandException) if ex.getErrorMessage.endsWith("not found.") || ex.getErrorCode == 26 =>
        logInfo(s"Could not find collection (${readConfig.collectionName}), using a single partition")
        MongoSinglePartitioner.partitions(connector, readConfig, pipeline)
      case Failure(e) =>
        logWarning(s"Could not get collection statistics. Server errmsg: ${e.getMessage}")
        throw e
    }
  }
  // scalastyle:on cyclomatic.complexity
}

　　　　上面这段是connector中的代码，作用是划分partition，sample的大小不会影响数据准确性，只会影响性能。改成1可以避免这个错误。

2、带验证的数据读写，不能初始化类。

　　有些时候我会将部分数据读出来，修改后更新回原collection，但是MongoSpark.write操作只能写入到一张不存在的表中，不能进行upsert，所以需要在分发任务的时候序列化一个(val dbColl: MongoCollection[Document] = db.getCollection(""))，把它传到每个task中，这样我就可以处理完一部分数据之后直接用这个MongoCollection写回到表里，rdd.foreach(x => {????  dbColl.replaceOne(eqq("_id", y.get("_id")), y, new UpdateOptions().upsert(true))})。这个在没有认证的时候是可以的，但是在有认证的情况下会报不能初始化类的错误，仔细看错误路径发现报错部分在dbColl.replaceOne这句。经过若干次尝试，发现：

　　a) 如果在rdd.foreach之前使用val mongoURI = new MongoClientURI(uri)会报MongoClientURI不能被序列化的错。

　　b) 如果这个写在main()之前，会出现不能初始化类的错误。

　　c) 如果这个写在里面，但是rdd.foreach中用到了外面声明的一个Calendar，也会报不能初始化类的错误。

　　d) 把Calendar转化为字符串，val mongoURI = new MongoClientURI(uri)都写在foreach里面，可以正确执行。

　　得出结论如下：

　　好多乱七八糟的类不能被序列化，不能被序列化的东西需要分发的话就会报这个错误。但是知道了原理还得想个解决方案，毕竟每条数据都创建一个mongoclient实在有点怪怪的，所以最后决定使用foreachPartition，具体如下：

  	val rdd = MongoSpark.builder().sparkSession(spark).build().toRDD()
//  	val df = MongoSpark.builder().sparkSession(spark).build.toDF
//  	MongoSpark.save(df.write)
  	rdd.cache()
   	println(rdd.count())  
   	val uri = s"mongodb://${user}:${passwd}@192.168.12.161:27017/?authSource=${authDB}"
   	val uri2 = s"mongodb://${config.getString("mongo.backup.user")}:${config.getString("mongo.backup.passwd")}@192.168.12.160:27017/?authSource=${config.getString("mongo.backup.authDB")}"
  	rdd.foreachPartition { x => {  		
  		val mongoURI = new MongoClientURI(uri)
			val mongo = new MongoClient(mongoURI)
			val db = mongo.getDatabase("wenshu")
			val dbColl = db.getCollection("origin")
			
			//val mongoURI2 = new MongoClientURI(s"mongodb://${config.getString("mongo.backup.user")}:${config.getString("mongo.backup.user")}@192.168.12.160:27017/?authSource=${config.getString("mongo.backup.user")}")
			val mongoURI2 = new MongoClientURI(uri2)
			val mongo2 = new MongoClient(mongoURI2)
			val db2 = mongo2.getDatabase("wenshu")
			val dbColl2 = db2.getCollection(backName)

			x.foreach { y => 
				dbColl.replaceOne(eqq("_id", y.get("_id")), y, new UpdateOptions().upsert(true))
				dbColl2.insertOne(y)
  		}
  		mongo.close
  		mongo2.close
  	} }