i am trying to establish the connection to Apache Solr from Apache Spark Dataframe and retrieve records. but getting Null Pointer exception while loading Solr collection to Spark Dataframe.
I am able to load Apache solr index in a Spark Dataframe separate program . But when i try to add/integrate the solr connection logic( val options = Map("collection" -> "entity_APPLICATION", "zkhost" -> "localhost/solr")
val initsolrdf = sqlContext.read.format("solr").options(options).load into SparkStreaming flow , i am getting null pointer exception while connecting to Solr from dataframe.
val options = Map("collection" -> "lily_entity_APPLICATION", "zkhost" -> "localhost/solr")
val initsolrdf = sqlContext.read.format("solr").options(options).load initsolrdf.show()
I am working on consuming messages from kafka and getting few fields with values and those fields i need to send to solr as filter to get require indexed values from Solr.So my code consumed kafka message and in the same code while initiating connection to Solr via Apache Spark Dataframe , i am getting error .
When i try the connection settings in different program in the same project its working fine.
Working connection :
val initsolrdf = sqlContext.load("solr", Map("zkHost" -> "localhost/solr", "collection" -> "entity_APPLICATION")) initsolrdf.printSchema initsolrdf.registerTempTable("ProducrDeDupe")
Below is the part of main code where i am consuming json and retrieving few fields with values to send as filter to solr index via Apache Spark dataframe.
def parseJson(jsonString: String): String = {
val json: Option[Any] = JSON.parseFull(jsonString);
val jsonMap = json.get.asInstanceOf[Map[String, Any]];
val application_processing = jsonMap.get("application-processing").get.asInstanceOf[Map[String, String]];
val applicant = jsonMap.get("applicant").get.asInstanceOf[Map[String, String]];
val application = jsonMap.get("application").get.asInstanceOf[Map[String, String]];
val product = jsonMap.get("product").get.asInstanceOf[Map[String, String]];
val product = jsonMap.get("product").get.asInstanceOf[Map[String, String]];
val product_category = product.get("product-category").get.asInstanceOf[String];
val product_type = product.get("product-type").get.asInstanceOf[String];
val fullname = applicant.get("full-name").get.asInstanceOf[String];
val owner_id = application.get("owner-id").get.asInstanceOf[String];
val duplicate_flag = application_processing.get("duplicate-flag").get.asInstanceOf[Boolean];
I have not passed any columns with values to Solr as a filters but just trying to load solr collection into Spark dataframe below.
/* Initiating connection to Apache Solr and checking any dupliacte product type already created for the Applicant */
val options = Map("collection" -> "lily_entity_APPLICATION", "zkhost" -> "localhost/solr")
val initsolrdf = sqlContext.read.format("solr").options(options).load
initsolrdf.show()
if (initsolrdf.count() != 0) {
flag = "Y";
} else {
flag = "N";[enter image description here][1]
}
val flag_update = application_processing + ("duplicate_flag" -> flag); val updatedJson = jsonMap + ("application-processing" -> flag_update); // } I have atatched the error message screenshot . Could you please assist where i go wrong .
... View more
