Время Spark при параллельном выполнении в Spark для вызова API

Question

Я делаю это ниже на моем 8-гигабайтном ноутбуке и запускаю код в Intellij.Я вызываю 3 apis параллельно с функцией map и библиотекой scalaj и вычисляю время вызова каждого API следующим образом:

val urls = spark.sparkContext.parallelize(Seq("url1", "url2", "url3"))
//for each API call,execute them in different executor and collate data 
val actual_data = urls.map(x => spark.time(HTTPRequestParallel.ds(x))) 

Когда выполняется spark.time, я ожидал 3 набора временино это дает мне 6 наборов времени

Time taken: 14945 ms
Time taken: 21773 ms
Time taken: 22446 ms
Time taken: 6438 ms
Time taken: 6877 ms
Time taken: 7107 ms

Что мне здесь не хватает и действительно ли это параллельные вызовы API в природе?

Answer 1

На самом деле, один только фрагмент кода вообще не будет выполнять spark.time, функция map является ленивой, поэтому она не будет выполняться, пока вы не выполните действие с RDD.Вам также следует учитывать, что если вы не сохраните свой преобразованный RDD, он пересчитает все преобразования для каждого действия.Это означает, что если вы делаете что-то вроде этого:

val urls = spark.sparkContext.parallelize(Seq("url1", "url2", "url3"))
//for each API call,execute them in different executor and collate data 
val actual_data = urls.map(x => spark.time(HTTPRequestParallel.ds(x))) 
val c = actual_data.count()
actual_data.collect()

Будет 6 выполнений того, что определено внутри map (два для каждого элемента в RDD, первый дляcount и второе для collect).Чтобы избежать этого повторного вычисления, вы можете кэшировать или сохранить RDD следующим образом

val urls = spark.sparkContext.parallelize(Seq("url1", "url2", "url3"))
//for each API call,execute them in different executor and collate data 
val actual_data = urls.map(x => spark.time(HTTPRequestParallel.ds(x))).cache() 
val c = actual_data.count()
actual_data.collect()

Во втором примере вы увидите только 3 журнала вместо 6