I want to query MongoDB collection with Spark using PySpark. But when I use "collect", "count", etc. with some filter conditions(for examle field > 5) the job doesn't appear for long time on Spark interface and takes too long to complete. But in small collections there is no problem, everything goes normal. I think Apache Spark tries to read all data.
My confs;
conf = SparkConf()\
.set('spark.executor.memory', '7g')\
.set('spark.executor.cores', '2')\
.set('spark.num.executors', '8')\
.set('spark.cores.max', '16')\
.set('spark.driver.memory', '10g')\
.set('spark.driver.port', '10001')\
.set('spark.driver.host', 'xx.xx.xx.xx')\
.set('spark.dynamicAllocation.enabled', "false")\
.set("spark.mongodb.read.connection.uri", "mongodb://XX.XX.XX.XX:27017")\
.set("spark.mongodb.read.database", "collection_1")\
.set("spark.mongodb.read.collection", "ADPacket")\
.set("spark.mongodb.write.connection.uri", "mongodb://XX.XX.XX.XX:27017")\
.set("spark.mongodb.write.database", "collection_1")\
.set("spark.mongodb.write.collection", "ADPacket")\
.set("spark.mongodb.write.operationType", "update")\
.set("spark.mongodb.readPreference.name", "secondaryPreferred")\
.setAppName('hello')\
.setMaster('spark://xx.xx.xx.xx:7077')
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")
sqlContext = SQLContext(sc)
and I load the data;
df = sqlContext.read.format("mongodb").load()
df.createOrReplaceTempView("ADPacket")
I tried both below commands.
sqlContext.sql("SELECT * FROM ADPacket WHERE AssetConnectDeviceKey='xxxx' and TimeStamp<1655555989 LIMIT 500").collect()
df.filter((df["TimeStamp"]>1650050989) & (df["TimeStamp"]<1655555989)).count()
In addition AssetConnectDeviceKey and TimeStamp are indexes in MongoDB collection.