代码:
import com.mongodb.spark.config.ReadConfig import com.mongodb.spark.sql._ val config = sqlContext.sparkContext.getConf .set("spark.mongodb.keep_alive_ms", "15000") .set("spark.mongodb.input.uri", "mongodb://10.100.12.14:27017") .set("spark.mongodb.input.database", "bi") .set("spark.mongodb.input.collection", "userGroupMapping") val readConfig = ReadConfig(config) val objUserGroupMapping = sqlContext.read .format("com.mongodb.spark.sql") .mongo(readConfig) objUserGroupMapping.printSchema() val tbUserGroupMapping=objUserGroupMapping.toDF() tbUserGroupMapping.registerTempTable("userGroupMapping") select _id,c,g,n,rn,t,ut from userGroupMapping where ut>‘2018-05-02‘ limit 100
使用上述的代码取userGroupMapping 集合过后的100条记录,出现了executor time out 和GC overhead limit exceeded 异常。一开始以为推测是task 从mongodb取的数据太大,导致spark executor内存不够,后来调研了一下spark mongodb connector 在取数据的时候是条件下发的,也就是先从mongodb过滤再取回spark 内存,这样的话就不会出现内存不够的情况。后来在网上调研后,有一种说法是说task 太多,导致task gc的时候争夺gc时间和内存资源(这个不是很清楚),根据这种说法,我把本来task core 从16个调成6个后再跑程序,居然不会报错了。至于 具体原因还不是很清楚,先在这里记录一下。
原文地址:https://www.cnblogs.com/chengjunhao/p/8984466.html
时间: 2024-11-08 19:17:15