１.$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
} }