Спарк Scala Получить данные от rdd.foreachPartition

Question

У меня есть некоторый код, как это:

 println("\nBEGIN Last Revs Class: "+ distinctFileGidsRDD.getClass) 
     val lastRevs = distinctFileGidsRDD. 
     foreachPartition(iter => { 
      SetupJDBC(jdbcDriver, jdbcUrl, jdbcUser, jdbcPassword) 
      while(iter.hasNext) { 
      val item = iter.next() 
      //println(item(0)) 
      println("String: "+item(0).toString()) 
      val jsonStr = DB.readOnly { implicit session => 
       sql"SELECT jsonStr FROM lasttail WHERE fileGId = ${item(0)}::varchar". 
       map { resultSet => resultSet.string(1) }.single.apply() 
      } 
      println("\nJSON: "+jsonStr) 
      } 
     }) 
     println("\nEND Last Revs Class: "+ lastRevs.getClass) 

кода выводит (с тяжелыми правок) что-то вроде:

BEGIN Last Revs Class: class org.apache.spark.rdd.MapPartitionsRDD 
String: 1fqhSXPE3GwrJ6SZzC65gJnBaB5_b7j3pWNSfqzU5FoM 
JSON: Some({"Struct":{"fileGid":"1fqhSXPE3GwrJ6SZzC65gJnBaB5_b7j3pWNSfqzU5FoM",...) 
String: 1eY2wxoVq17KGMUBzCZZ34J9gSNzF038grf5RP38DUxw 
JSON: Some({"Struct":{"fileGid":"1fqhSXPE3GwrJ6SZzC65gJnBaB5_b7j3pWNSfqzU5FoM",...) 
... 
JSON: None() 
END Last Revs Class: void 

ВОПРОС 1: Как можно Я получаю значение lastRevs в полезном формате, таком как строка/нуль JSON или параметр Some/None?

ВОПРОС 2: Мое предпочтение: есть ли другой способ получить данные разделов, подобные RDD-формату (а не формат итератора)?

dstream.foreachRDD { (rdd, time) => 
    rdd.foreachPartition { partitionIterator => 
    val partitionId = TaskContext.get.partitionId() 
    val uniqueId = generateUniqueId(time.milliseconds, partitionId) 
    // use this uniqueId to transactionally commit the data in partitionIterator 
    } 
} 

из http://spark.apache.org/docs/latest/streaming-programming-guide.html#performance-tuning

ВОПРОС 3: Является ли метод получения данных, которые я использую в здравом уме метод (с учетом я следую по ссылке выше)? (Отложите в сторону тот факт, что это сейчас система JDBC scalikejdbc. Это будет ключ, хранилище значений другого типа, кроме этого прототипа.)

Answer 1

Чтобы создать преобразование, использующее ресурсы, локальные для исполнителя (например, DB или сетевое соединение), вы должны использовать rdd.mapPartitions. Он позволяет инициализировать код локально исполнителю и использовать эти локальные ресурсы для обработки данных в разделе.

Код должен выглядеть следующим образом:

val lastRevs = distinctFileGidsRDD. 
     mapPartitions{iter => 
      SetupJDBC(jdbcDriver, jdbcUrl, jdbcUser, jdbcPassword) 
      iter.map{ element => 
      DB.readOnly { implicit session => 
       sql"SELECT jsonStr FROM lasttail WHERE fileGId = ${element(0)}::varchar" 
       .map { resultSet => resultSet.string(1) }.single.apply() 
      } 
      } 
     }