py4j.protocol.Py4JJavaError: Произошла ошибка при вызове o788.save. : com.mongodb.MongoTimeoutException, WritableServerSelector

Question

Версия Pyspark: 2.4.4 Версия MongoDB: 4.2.0. ОЗУ: 64 ГБ. Ядро ЦП: 32: работает скрипт: spark-submit --executor-memory 8G --driver-memory 8G - пакеты org.mongodb.spark: mongo-spark-connector_2.11: 2.3.1 demographic.py

, когда я запускаю код, я получаю сообщение об ошибке: "py4j.protocol.Py4JJavaError: Произошла ошибка при вызове o764.save.: com.mongodb.MongoTimeoutException: Превышено время ожидания после 30000 мс в ожидании сервера, который соответствует WritableServerSelector. Клиентское представление состояния кластера: {type = REPLICA_SET, servers = [{address = 172. . . *: 27017, type =REPLICA_SET_SECONDARY, roundTripTime = 34,3 мс, состояние = CONNECTED}] "

Я пытаюсь прочитать коллекцию MongoDB с одного сервера реплики, который имеет аутентификацию, и я могу читать с этого сервера, используя:

df_ipapp = spark.read.format('com.mongodb.spark.sql.DefaultSource').option('uri', '{}/{}.IpAppointment?authSource={}'.format(mongo_url, mongo_db,auth_source)).load()

и работает нормально. но после обработки этого фрейма данных я записываю этот фрейм данных в другую MongoDB, которая не имеет аутентификации, которая находится локально в месте, где я обрабатываю, используя: df.write.format ('com.mongodb.spark.sql.DefaultSource'). mode ('overwrite'). option ('uri', '{} / {}. demographic'.format (mongo_final_url, mongo_final_db)). save ()

и каждый раз, когда я получаю ошибку здесь

  File "/home/svr_data_analytic/hmis-analytics-data-processing/src/main/python/sales/demographic.py", line 297, in save_n_rename
    .option('uri', '{}/{}.demographic'.format(mongo_url, mongo_final_db)).save()
  File "/home/svr_data_analytic/spark/spark-2.4.4-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 736, in save
  File "/home/svr_data_analytic/spark/spark-2.4.4-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
  File "/home/svr_data_analytic/spark/spark-2.4.4-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/sql/utils.py", line 63, in deco
  File "/home/svr_data_analytic/spark/spark-2.4.4-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o788.save.
: com.mongodb.MongoTimeoutException: Timed out after 30000 ms while waiting for a server that matches WritableServerSelector. Client view of cluster state is {type=REPLICA_SET, servers=[{address=172.*.*.*:27017, type=REPLICA_SET_SECONDARY, roundTripTime=0.8 ms, state=CONNECTED}]

чтение с сервера реплики:

df_bills = spark.read.format('com.mongodb.spark.sql.DefaultSource').option('uri', '{}/{}.Bills?authSource={}'.format(mongo_url, mongo_db, auth_source)).load()

запись в mongodb:

df.write.format('com.mongodb.spark.sql.DefaultSource').mode('overwrite').option('uri', '{}/{}.demographic'.format(mongo_final_url, mongo_final_db)).save()

Я хочу прочитать с сервера реплики MondoDb, который имеет аутентификацию, обработать фрейм данных и записатьэто местному MongoDB заранее спасибо