I worked with Learning Spark chapter 6. working with partitions.
But, I had JsonMappingException: No suitable constructor found for type [simple type, class $iwC$$iwC$CallLog]: can not instantiate from JSON object (need to add/enable type information?) CallLog class were defined as follows: case class CallLog(callsign: String="", contactlat: Double,
contactlong: Double, mylat: Double, mylong: Double) http response data like this: http://new73s.herokuapp.com/qsos/KK6JKQ.json My code gererating Exception is: import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import org.eclipse.jetty.client.ContentExchange
import org.eclipse.jetty.client.HttpClient
// define calllog format - real data types are all String.
case class CallLog(callsign: String="", contactlat: Double,
contactlong: Double, mylat: Double, mylong: Double)
// create JSON readerWriter
def createMapper() = {
val mapper = new ObjectMapper()
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
mapper.registerModule(DefaultScalaModule)
mapper
}
// http container which delivers (JSON typed)calllog per callsign.
def createExchangeForSign(client: HttpClient, sign: String): (String, ContentExchange) = {
val exchange = new ContentExchange()
exchange.setURL(s"http://new73s.herokuapp.com/qsos/${sign}.json")
client.send(exchange)
(sign, exchange)
}
// Draw out json typed response,
// Read it with CallLog format, and Return Array[CallLog].
def readExchangeCallLog(mapper: ObjectMapper, exchange: ContentExchange): Array[CallLog] = {
exchange.waitForDone()
val responseJson = exchange.getResponseContent()
val qsos = mapper.readValue(responseJson, classOf[Array[CallLog]])
qsos
}
val contactsContactLists = validSigns.distinct().mapPartitions{
signs =>
val mapper = createMapper()
// create a connection pool
val client = new HttpClient()
client.start()
// create http request
signs.map {sign =>
createExchangeForSign(client, sign)
// fetch responses
}.map{ case (sign, exchange) =>
(sign, readExchangeCallLog(mapper, exchange))
}.filter(x => x._2 != null) // Remove empty CallLogs
}
println(contactsContactLists.collect().toList) validSigns RDD have lazy data liky [u'W8PAL', u'KK6JKQ', u'W6BB', u'VE3UOW', u'VE2CUA', u'VE2UN', u'OH2TI', u'GB1MIR', u'K2AMH', u'UA1LO', u'N7ICE']. How can I try to remove the Exception and get results.
I worked with zeppelin at HDP 2.5 sandbox.
... View more
