Error : com.mongodb.MongoCursorNotFoundException : MongoDB에서 Spark로 읽는 중

Question

2 억 3 천 4 백만 레코드로 Spark에서 MongoDB의 모음을 읽으려고합니다. 1 필드 만 원한다.

case class Linkedin_Profile(experience : Array[Experience]) 
case class Experience(company : String) 

val rdd = MongoSpark.load(sc, ReadConfig(Map("uri" -> mongo_uri_linkedin))) 
val company_DS = rdd.toDS[Linkedin_Profile]() 
val count_udf = udf((x: scala.collection.mutable.WrappedArray[String]) => {x.filter(_ != null).groupBy(identity).mapValues(_.size)}) 
val company_ColCount = company_DS.select(explode(count_udf($"experience.company"))) 
comp_rdd.saveAsTextFile("/dbfs/FileStore/chandan/intermediate_count_results.csv") 

작업이 완료된 작업의 절반 1 시간 동안 실행하지만 오류를

com.mongodb.MongoCursorNotFoundException: 
Query failed with error code -5 and error message 
'Cursor 8962537864706894243 not found on server cluster0-shard-01-00-i7t2t.mongodb.net:37017' 
on server cluster0-shard-01-00-i7t2t.mongodb.net:37017 

을 제공합니다 그 후 나는 아래에 구성을 변경했지만, 아무 소용.

System.setProperty("spark.mongodb.keep_alive_ms", "7200000") 

큰 컬렉션을 읽는 방법을 제안하십시오.

Answer 1

config 속성 park.mongodb.keep_alive_ms은 클라이언트의 수명을 제어하기위한 것입니다. docs here을 참조하십시오.

발생하는 문제는 서버 쪽 구성과 관련이있는 것으로 보입니다. on this issue을 문서화 무슨에 따르면

By specifing the cursorTimeoutMillis option, administrators can configure mongod or mongos to automatically remove idle client cursors after a specified interval. The timeout applies to all cursors maintained on a mongod or mongos, may be specified when starting the mongod or mongos and may be modified at any time using the setParameter command.

그래서, 같은 지정된 cursorTimeoutMillis로 mongod 데몬을 시작하십시오 :

mongod --setParameter cursorTimeoutMillis=10800000 

이 명령은 3 시간 동안 유효한 커서를 유지하는 서버에 지시하려고합니다.

이론적으로는 성가심을 없앨 수 있지만 읽기를 더 빨리 완료하는 것이 좋습니다. 컬렉션에있는 데이터 세트를 Spark에 실제로로드하려는 것으로 제한 할 수 있습니다. 살펴볼 가치가있는 읽기 속도를 조정할 수있는 많은 옵션이있을 수 있습니다.