spark connects to mongoDB sharded cluster, but no data is fetched

1
votes

Environment:

  1. Four Debian 9 servers (named visa0, visa1, visa2, visa3)
  2. Spark (v2.4.0) cluster on 4 nodes (visa1: master, visa0..3: slaves)
  3. MongoDB (v3.2.11) sharded cluster con 4 nodes ( config server replica set on visa1..3, mongos on visa1, shard servers: visa0..3 )
  4. I'm using Spark MongoDB connector installed with "spark-shell --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0"
  5. and Jupyter Notebook, Python 3 (pyspark v. 2.4.0)

Problem:

I can create a SparkSession connected to the master, and load a DataFrame with the whole content of a Mongo collection. In fact, I get the DataFrame schema correctly. But, with .count() or .show() methods on the dataframe i get 0 results.

Python/pyspark code:

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--driver-memory 6g --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0 pyspark-shell'

import pyspark
sparkSession = pyspark.sql.SparkSession \
  .builder \
  .master('spark://visa1:7077') \
  .appName("myApp") \
  .config("spark.executor.memory", "4g") \
  .config("spark.mongodb.input.uri", "mongodb://visa1/email.emails") \
  .config("spark.mongodb.input.partitioner" ,"MongoShardedPartitioner") \
  .getOrCreate()

df = sparkSession.read.format("com.mongodb.spark.sql.DefaultSource") \
  .option("uri", "mongodb://visa1/email.emails").load()

df.printSchema()
# gets the schema correctly

df.count()
# gets 0, when there are more than 750.000 documents on the collection

Considerations:

  • The same test connecting with the same code to a standalone mongo server works fine (df.count() gives the correct count).
  • connecting to mongos, db.emails.count() gives the correct count
  • Config Server Replica Set seems ok (through rs.status() command on primary server)
  • Sharding seems ok (through sh.status() command on mongos)

  • on spark executors, i get the following on stderr:

    Spark Executor Command: "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java" "-cp" "/root/spark/conf/:/root/spark/jars/*" "-Xmx4096M" "-Dspark.driver.port=36511" "org.apache.spark.executor.CoarseGrainedExecutorBackend" "--driver-url" "spark://CoarseGrainedScheduler@visa1:36511" "--executor-id" "2" "--hostname" "visa2" "--cores" "6" "--app-id" "app-20190106213435-0003" "--worker-url" "spark://Worker@visa2:46705"

  • on spark executors, i get the following on stdout (please, note the "cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out"):

    2019-01-06 21:34:35 INFO  CoarseGrainedExecutorBackend:2566 - Started daemon with process name: 18812@visa2
2019-01-06 21:34:35 INFO  SignalUtils:54 - Registered signal handler for TERM
2019-01-06 21:34:35 INFO  SignalUtils:54 - Registered signal handler for HUP
2019-01-06 21:34:35 INFO  SignalUtils:54 - Registered signal handler for INT
2019-01-06 21:34:36 WARN  NativeCodeLoader:62 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2019-01-06 21:34:36 INFO  SecurityManager:54 - Changing view acls to: root
2019-01-06 21:34:36 INFO  SecurityManager:54 - Changing modify acls to: root
2019-01-06 21:34:36 INFO  SecurityManager:54 - Changing view acls groups to: 
2019-01-06 21:34:36 INFO  SecurityManager:54 - Changing modify acls groups to: 
2019-01-06 21:34:36 INFO  SecurityManager:54 - SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(root); groups with view permissions: Set(); users  with modify permissions: Set(root); groups with modify permissions: Set()
2019-01-06 21:34:37 INFO  TransportClientFactory:267 - Successfully created connection to visa1/1.1.241.71:36511 after 103 ms (0 ms spent in bootstraps)
2019-01-06 21:34:37 INFO  SecurityManager:54 - Changing view acls to: root
2019-01-06 21:34:37 INFO  SecurityManager:54 - Changing modify acls to: root
2019-01-06 21:34:37 INFO  SecurityManager:54 - Changing view acls groups to: 
2019-01-06 21:34:37 INFO  SecurityManager:54 - Changing modify acls groups to: 
2019-01-06 21:34:37 INFO  SecurityManager:54 - SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(root); groups with view permissions: Set(); users  with modify permissions: Set(root); groups with modify permissions: Set()
2019-01-06 21:34:37 INFO  TransportClientFactory:267 - Successfully created connection to visa1/1.1.241.71:36511 after 2 ms (0 ms spent in bootstraps)
2019-01-06 21:34:37 INFO  DiskBlockManager:54 - Created local directory at /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/blockmgr-411ce01c-f631-45b5-9b60-b7d6c124d289
2019-01-06 21:34:37 INFO  MemoryStore:54 - MemoryStore started with capacity 2004.6 MB
2019-01-06 21:34:37 INFO  CoarseGrainedExecutorBackend:54 - Connecting to driver: spark://CoarseGrainedScheduler@visa1:36511
2019-01-06 21:34:37 INFO  WorkerWatcher:54 - Connecting to worker spark://[email protected]:46705
2019-01-06 21:34:37 INFO  TransportClientFactory:267 - Successfully created connection to /1.1.237.142:46705 after 2 ms (0 ms spent in bootstraps)
2019-01-06 21:34:37 INFO  WorkerWatcher:54 - Successfully connected to spark://[email protected]:46705
2019-01-06 21:34:37 INFO  CoarseGrainedExecutorBackend:54 - Successfully registered with driver
2019-01-06 21:34:37 INFO  Executor:54 - Starting executor ID 2 on host 1.1.237.142
2019-01-06 21:34:37 INFO  Utils:54 - Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 44735.
2019-01-06 21:34:37 INFO  NettyBlockTransferService:54 - Server created on 1.1.237.142:44735
2019-01-06 21:34:37 INFO  BlockManager:54 - Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
2019-01-06 21:34:37 INFO  BlockManagerMaster:54 - Registering BlockManager BlockManagerId(2, 1.1.237.142, 44735, None)
2019-01-06 21:34:37 INFO  BlockManagerMaster:54 - Registered BlockManager BlockManagerId(2, 1.1.237.142, 44735, None)
2019-01-06 21:34:37 INFO  BlockManager:54 - Initialized BlockManager: BlockManagerId(2, 1.1.237.142, 44735, None)
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 1
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 5
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 9
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 13
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 17
2019-01-06 21:35:17 INFO  CoarseGrainedExecutorBackend:54 - Got assigned task 21
2019-01-06 21:35:17 INFO  Executor:54 - Running task 16.0 in stage 1.0 (TID 17)
2019-01-06 21:35:17 INFO  Executor:54 - Running task 8.0 in stage 1.0 (TID 9)
2019-01-06 21:35:17 INFO  Executor:54 - Running task 0.0 in stage 1.0 (TID 1)
2019-01-06 21:35:17 INFO  Executor:54 - Running task 20.0 in stage 1.0 (TID 21)
2019-01-06 21:35:17 INFO  Executor:54 - Running task 4.0 in stage 1.0 (TID 5)
2019-01-06 21:35:17 INFO  Executor:54 - Running task 12.0 in stage 1.0 (TID 13)
2019-01-06 21:35:17 INFO  Executor:54 - Fetching spark://visa1:36511/files/org.mongodb_mongo-java-driver-3.9.0.jar with timestamp 1546806874832
2019-01-06 21:35:17 INFO  TransportClientFactory:267 - Successfully created connection to visa1/1.1.241.71:36511 after 5 ms (0 ms spent in bootstraps)
2019-01-06 21:35:17 INFO  Utils:54 - Fetching spark://visa1:36511/files/org.mongodb_mongo-java-driver-3.9.0.jar to /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/fetchFileTemp6501978500036245382.tmp
2019-01-06 21:35:18 INFO  Utils:54 - Copying /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/-13359565311546806874832_cache to /root/spark/work/app-20190106213435-0003/2/./org.mongodb_mongo-java-driver-3.9.0.jar
2019-01-06 21:35:18 INFO  Executor:54 - Fetching spark://visa1:36511/files/org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar with timestamp 1546806874820
2019-01-06 21:35:18 INFO  Utils:54 - Fetching spark://visa1:36511/files/org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar to /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/fetchFileTemp205676444589226484.tmp
2019-01-06 21:35:18 INFO  Utils:54 - Copying /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/8587355671546806874820_cache to /root/spark/work/app-20190106213435-0003/2/./org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar
2019-01-06 21:35:18 INFO  Executor:54 - Fetching spark://visa1:36511/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar with timestamp 1546806874797
2019-01-06 21:35:18 INFO  Utils:54 - Fetching spark://visa1:36511/jars/org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar to /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/fetchFileTemp2003659413222858965.tmp
2019-01-06 21:35:18 INFO  Utils:54 - /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/-10843728141546806874797_cache has been previously copied to /root/spark/work/app-20190106213435-0003/2/./org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar
2019-01-06 21:35:18 INFO  Executor:54 - Adding file:/root/spark/work/app-20190106213435-0003/2/./org.mongodb.spark_mongo-spark-connector_2.11-2.4.0.jar to class loader
2019-01-06 21:35:18 INFO  Executor:54 - Fetching spark://visa1:36511/jars/org.mongodb_mongo-java-driver-3.9.0.jar with timestamp 1546806874798
2019-01-06 21:35:18 INFO  Utils:54 - Fetching spark://visa1:36511/jars/org.mongodb_mongo-java-driver-3.9.0.jar to /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/fetchFileTemp1454666184402659399.tmp
2019-01-06 21:35:18 INFO  Utils:54 - /tmp/spark-ae02f35d-8340-4cda-ba6f-8d8b7138e803/executor-b6a0e407-de5a-420a-a528-96573fcd9700/spark-1f05e532-25da-492b-8e52-3e5da3fd9617/20228089061546806874798_cache has been previously copied to /root/spark/work/app-20190106213435-0003/2/./org.mongodb_mongo-java-driver-3.9.0.jar
2019-01-06 21:35:18 INFO  Executor:54 - Adding file:/root/spark/work/app-20190106213435-0003/2/./org.mongodb_mongo-java-driver-3.9.0.jar to class loader
2019-01-06 21:35:18 INFO  TorrentBroadcast:54 - Started reading broadcast variable 2
2019-01-06 21:35:18 INFO  TransportClientFactory:267 - Successfully created connection to /1.1.241.71:38095 after 4 ms (0 ms spent in bootstraps)
2019-01-06 21:35:18 INFO  MemoryStore:54 - Block broadcast_2_piece0 stored as bytes in memory (estimated size 7.5 KB, free 2004.6 MB)
2019-01-06 21:35:18 INFO  TorrentBroadcast:54 - Reading broadcast variable 2 took 182 ms
2019-01-06 21:35:18 INFO  MemoryStore:54 - Block broadcast_2 stored as values in memory (estimated size 15.8 KB, free 2004.6 MB)
2019-01-06 21:35:18 INFO  TorrentBroadcast:54 - Started reading broadcast variable 0
2019-01-06 21:35:18 INFO  MemoryStore:54 - Block broadcast_0_piece0 stored as bytes in memory (estimated size 396.0 B, free 2004.6 MB)
2019-01-06 21:35:18 INFO  TorrentBroadcast:54 - Reading broadcast variable 0 took 14 ms
2019-01-06 21:35:18 INFO  MemoryStore:54 - Block broadcast_0 stored as values in memory (estimated size 200.0 B, free 2004.6 MB)
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster created with settings {hosts=[visa1:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500}
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  cluster:71 - Cluster description not yet available. Waiting for 30000 ms before timing out
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:5}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:3}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:6}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:1}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:4}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:2}] to visa1:27017
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=2389159}
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=3296820}
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=3158622}
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=2556701}
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=2174393}
2019-01-06 21:35:19 INFO  cluster:71 - Monitor thread successfully connected to server with description ServerDescription{address=visa1:27017, type=SHARD_ROUTER, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 2, 11]}, minWireVersion=0, maxWireVersion=4, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=null, roundTripTimeNanos=7550692}
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Creating MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:11}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:9}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:12}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:8}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:10}] to visa1:27017
2019-01-06 21:35:19 INFO  connection:71 - Opened connection [connectionId{localValue:7}] to visa1:27017
2019-01-06 21:35:19 INFO  CodeGenerator:54 - Code generated in 259.273212 ms
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 12.0 in stage 1.0 (TID 13). 1586 bytes result sent to driver
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 8.0 in stage 1.0 (TID 9). 1586 bytes result sent to driver
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 16.0 in stage 1.0 (TID 17). 1586 bytes result sent to driver
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 0.0 in stage 1.0 (TID 1). 1586 bytes result sent to driver
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 20.0 in stage 1.0 (TID 21). 1586 bytes result sent to driver
2019-01-06 21:35:20 INFO  Executor:54 - Finished task 4.0 in stage 1.0 (TID 5). 1586 bytes result sent to driver
2019-01-06 21:35:25 INFO  MongoClientCache:48 - Closing MongoClient: [visa1:27017]
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:9}] to visa1:27017 because the pool has been closed.
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:7}] to visa1:27017 because the pool has been closed.
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:10}] to visa1:27017 because the pool has been closed.
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:11}] to visa1:27017 because the pool has been closed.
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:12}] to visa1:27017 because the pool has been closed.
2019-01-06 21:35:25 INFO  connection:71 - Closed connection [connectionId{localValue:8}] to visa1:27017 because the pool has been closed.

Updated info (thanks to @kk1957 answer)

Making further tests, i'm pretty sure now that the problem comes from the SparkSession object initialization that is made in Jupyter Notebook:

  • when i start pyspark shell, all goes fine if I use the "spark" object created by pyspark
  • but, if i create a new SparkSession, i reproduce the lack of results.

Using default spark session:

./pyspark --master "spark://visa1:7077" --packages "org.mongodb.spark:mongo-spark-connector_2.11:2.4.0"

...

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Python version 3.5.3 (default, Sep 27 2018 17:25:39)
SparkSession available as 'spark'.
>>> df = spark.read.format("com.mongodb.spark.sql.DefaultSource") \
... .option("uri", "mongodb://visa1/email.emails") \
... .option("pipeline", '[ {"$match": {"mailbox": /^\/root\/pst_export\/albert_meyers_000_1_1.export/}} ]') \
... .load()
2019-01-07 08:41:30 WARN  Utils:66 - Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
>>> 
>>> df.count()
1162

But, creating my own spark session object:

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Python version 3.5.3 (default, Sep 27 2018 17:25:39)
SparkSession available as 'spark'.
>>> 
>>> spark2 = SparkSession \
...     .builder \
...     .master('spark://visa1:7077') \
...     .appName("myApp") \
...     .config("spark.executor.memory", "4g") \
...     .config("spark.mongodb.input.uri", "mongodb://visa1/email.emails") \
...     .config("spark.mongodb.input.partitioner" ,"MongoShardedPartitioner") \
...     .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.11:2.4.0") \
...     .getOrCreate()
>>> 
>>> df2 = spark2.read.format("com.mongodb.spark.sql.DefaultSource") \
... .option("uri", "mongodb://visa1/email.emails") \
... .option("pipeline", '[ {"$match": {"mailbox": /^\/root\/pst_export\/albert_meyers_000_1_1.export/}} ]') \
... .load()
2019-01-07 09:18:04 WARN  Utils:66 - Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
>>> 
>>> df2.count()
0

The same code, attacking a single MongoDB (no sharding) works fine:

./pyspark --master "spark://visa1:7077" --packages "org.mongodb.spark:mongo-spark-connector_2.11:2.4.0"

...

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Python version 3.5.3 (default, Sep 27 2018 17:25:39)
SparkSession available as 'spark'.
>>> 
>>> spark2 = SparkSession \
...                     .builder \
...                     .appName("myApp") \
...                     .config("spark.mongodb.input.uri", "mongodb://singleMongoDB/email.emails") \
...                     .config("spark.mongodb.input.partitioner" ,"MongoShardedPartitioner") \
...                     .getOrCreate()
>>> 
>>> df = spark2.read.format("com.mongodb.spark.sql.DefaultSource") \
... .option("uri", "mongodb://singleMongoDB/email.emails") \
... .option("pipeline", '[ {"$match": {"mailbox": /^\/root\/pst_export\/albert_meyers_000_1_1.export/}} ]') \
... .load()
2019-01-07 09:04:58 WARN  Utils:66 - Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.
>>> 
>>> df.count()
2019-01-07 09:05:03 WARN  MongoShardedPartitioner:60 - Collection 'email.emails' does not appear to be sharded, continuing with a single partition. To split the collections into multiple partitions connect to the MongoDB node directly
1162

Question:

I'm pretty sure the problem is on the way the SparkSession object is created in the Jupyter Notebook, when it attacks a MongoDB sharded cluster.

Could you help me to debug the problem?

Thanks in advance

1
mongodbapache-sparkpysparkshardingmongodb-cluster
The problem is with the partitioner used. See: stackoverflow.com/questions/54082209/spark-mongo-connector-mongoshardedpartitioner-does-not-work – jose

1 Answers

0
votes

A few suggestions:

1) Did you try connecting to Mongo db on the master machine? just to make sure there is nothing between the mongo and master.

2) Try running your cluster in a simpler configuration (without any executor or just one executor) and see if that helps you find the root cause.