Блог

Главная - Вопросы по программированию - Spark запускает только одного исполнителя при попытке чтения и записи данных в HDFS из Oracle table

Spark запускает только одного исполнителя при попытке чтения и записи данных в HDFS из Oracle table

#apache-spark

#apache-spark

Вопрос:

Я подключаюсь к oracle database JDBC connection использованию Spark и пытаюсь прочитать таблицу oracle, содержащую 40 миллионов строк. Я использую 30 исполнителей, 5 исполнительных ядер и память 4g для каждого исполнителя при запуске spark-shell / submit. При чтении счетчика или попытке записи данных фрейма данных для чтения / записи данных из oracle используется только один исполнитель. Попытался повторно разделить фрейм данных, но по-прежнему использовал только 1 исполнителя, что привело к значительному снижению производительности.

Ниже приведен используемый синтаксис, любое предложение приветствуется.

Фрагмент команды:-

 spark-shell --executor-memory 4G --executor-cores 5 --num-executors 30
val source_df = spark.read.format("jdbc").option("url", JDBC_URL).option("dbtable", src_table).option("user", *****).option("password", *****).option("driver", "oracle.jdbc.driver.OracleDriver").option("numPartitions", 40).option("partitionColumn", "*****").option("lowerBound", 1).option("upperBound", 100000).load()

val df_1_msag=source_table_DF_raw_msag.repartition(40)
df_1_msag.count

[Stage 0:=======================================================> (39   1

Комментарии:

1. Я предполагаю Stage-0 , что делает select count(*) , чтобы определить, как разделить ваш диапазон на разделы. Это будет сделано одной задачей.

Ответ №1:

Также важно количество одновременных подключений, разрешенных для пользователя в oracle db.

https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

numPartitions
Максимальное количество разделов, которые могут использоваться для параллелизма при чтении и записи таблиц. Это также определяет максимальное количество одновременных подключений JDBC. Если количество записываемых разделов превышает этот предел, мы уменьшаем его до этого предела, вызывая coalesce(numPartitions) перед записью.