Spark bean encoder for Java classes extending com.sun.jna.Structure

Viewed 408

I'm running some native code from within a Spark UDF using JNA. Currently I map those manually on Scala Case-Classes to be returned from the UDF (schema is inferred automatically). This leads to much code duplication, as the data-structures are almost identical.

So I declared a bunch of Java Classes to map the native datastructures (those classes need to inherit from com.sun.jna.Structure), they look like:

public class MyStruct extends com.sun.jna.Structure {

  public double[] cp_O = new double[3];
  public int B_HN;
  public double B_Yo;
  public double B_DY;

  @Override
  protected List getFieldOrder() {
    return Arrays.asList("cp_O","B_HN","B_Yo","B_DY");
}

}

Next, to use org.apache.spark.sql.Encoders.bean, I added getter and setter for this class (to make it a bean) and tried :

implicit val encoder = Encoders.bean(classOf[MyStruct])

// dummy udf without native call, just return a Bean
val myUDF = udf(() => new MyStruct(),encoder.schema)

ss.range(1)
  .withColumn("result",myUDF())
  .show()

This gives:

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:126) at org.apache.spark.sql.catalyst.JavaTypeInference$$anonfun$2.apply(JavaTypeInference.scala:125) 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:125) at org.apache.spark.sql.catalyst.JavaTypeInference$.inferDataType(JavaTypeInference.scala:55) at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.javaBean(ExpressionEncoder.scala:89) at org.apache.spark.sql.Encoders$.bean(Encoders.scala:142)

If I remove the extends Structure from my class, the code runs fine for spark, but not for JNA. So apparently Structure adds many fields to MyStruct, one of them are not supported/cannot be mapped. As I don't need those fields anyway, how can I formulate my code such that only "my" fields in MyStruct are encoded, but not those inherited from Structure?

0 Answers
Related