1.5.1| Spark Streaming | NullPointerException with SQL createDataFrame

Question

I am using Spark 1.5.1.

Within the streaming context I get the SQLContext as follows

SQLContext sqlContext = SQLContext.getOrCreate(records.context()); DataFrame dataFrame = sqlContext.createDataFrame(record, SchemaRecord.class); dataFrame.registerTempTable("records");

records is a JavaRDD Each Record has the following structure

public class SchemaRecord implements Serializable {

private static final long serialVersionUID = 1L; 
private String msisdn; 
private String application_type; 
//private long uplink_bytes = 0L;
}

When the field types like msisdn and application_type are only strings everything works fine.

The moment I add another field like uplink_bytes which is of Long type I get the following NullPointer Exception at createDataFrame

Exception in thread "main" java.lang.NullPointerException
at org.spark-project.guava.reflect.TypeToken.method(TypeToken.java:465)
at 
org.apache.spark.sql.catalyst.JavaTypeInference$$anonfun$2.apply(JavaTypeInference.scala:103)
at 
org.apache.spark.sql.catalyst.JavaTypeInference$$anonfun$2.apply(JavaTypeInference.scala:102)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108)
at org.apache.spark.sql.
catalyst.JavaTypeInference$.org$apache$spark$sql$catalyst$JavaTypeInference$$inferDataType(JavaTypeInference.scala:102)
at org.apache.spark.sql.catalyst.JavaTypeInference$.inferDataType(JavaTypeInference.scala:47)
at org.apache.spark.sql.SQLContext.getSchema(SQLContext.scala:1031)
at org.apache.spark.sql.SQLContext.createDataFrame(SQLContext.scala:519)
at org.apache.spark.sql.SQLContext.createDataFrame(SQLContext.scala:548)

Please suggest

Answer 1

Your problem is probably that you model class is not a clean JavaBean. At the moment Spark has no code to deal with properties that have a setter but no getter method. You can simply try something like this to inspect how Spark understands your class:

PropertyDescriptor[] props = Introspector.getBeanInfo(YourClass.class).getPropertyDescriptors();
for(PropertyDescriptor prop:props) {
    System.out.println(prop.getDisplayName());
    System.out.println("\t"+prop.getReadMethod());
    System.out.println("\t"+prop.getWriteMethod());
}

The introspector also recognizes fields that only have a setter as preoperties, which throws the NullPointerException in Spark.

Answer 2

Here is what I tried and it worked: -

Here is POJO which stores String, Long and Int Values: -

import java.io.*;

public class TestingSQLPerson implements Serializable {

// Here is Data in a comma Separated file: -
// Sumit,20,123455
// Ramit,40,12345

private String name;
private int age;
private Long testL;

public Long getTestL() {
    return testL;
}

public void setTestL(Long testL) {
    this.testL = testL;
}

public String getName() {
    return name;
}

public void setName(String name) {
    this.name = name;
}

public int getAge() {
    return age;
}

public void setAge(int age) {
    this.age = age;
}

}

And here is the Spark SQL Code in Java: -

import org.apache.spark.*;
import org.apache.spark.sql.*;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;

public class TestingLongSQLTypes {

public static void main(String[] args) {

    SparkConf javaConf = new SparkConf();
    javaConf.setAppName("Test Long TTyypes");
    JavaSparkContext javaCtx = new JavaSparkContext(javaConf);


    SQLContext sqlContext = new org.apache.spark.sql.SQLContext(javaCtx);

    String dataFile = "file:///home/ec2-user/softwares/crime-data/testfile.txt";
    JavaRDD<TestingSQLPerson> people = javaCtx.textFile(dataFile).map(
      new Function<String, TestingSQLPerson>() {
        public TestingSQLPerson call(String line) throws Exception {
          String[] parts = line.split(",");

          TestingSQLPerson person = new TestingSQLPerson();
          person.setName(parts[0]);
          person.setAge(Integer.parseInt(parts[1].trim()));
          person.setTestL(Long.parseLong(parts[2].trim()));

          return person;
        }
      });

    // Apply a schema to an RDD of JavaBeans and register it as a table.
    DataFrame schemaPeople = sqlContext.createDataFrame(people, TestingSQLPerson.class);
    schemaPeople.registerTempTable("TestingSQLPerson");

    schemaPeople.printSchema();
    schemaPeople.show();


}

}

All of the Above work and finally on the driver console i could see the results without any errors of Exception. @Yukti - In your case, It should work too, provided you follow the same steps as defined in the above examples. In case, there is any deviation, then let me know, I can try helping you.