java - 扩展 com.sun.jna.Structure 的 Java 类的 Spark bean 编码器

Question

我正在使用 JNA 从 Spark UDF 中运行一些本机代码。目前，我将这些手动映射到要从 UDF 返回的 Scala 案例类上（自动推断模式）。这会导致很多代码重复，因为数据结构几乎相同。

所以我声明了一堆 Java 类来映射本机数据结构（这些类需要继承自com.sun.jna.Structure），它们看起来像：

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");
}

}

接下来，为了使用org.apache.spark.sql.Encoders.bean，我为这个类添加了getter和setter（使其成为一个bean）并尝试：

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()

这给出了：

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 ) 在 scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 在 scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 在 scala.collection.mutable.ArrayOps$ofRef。 foreach(ArrayOps.scala:108) 在 scala.collection.TraversableLike$class.map(TraversableLike.scala:244) 在 scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108) 在 org.apache.spark.sql.catalyst.JavaTypeInference$.org$apache$spark$sql$catalyst$JavaTypeInference$$inferDataType(JavaTypeInference.scala:125) 在 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)

如果我extends Structure从我的班级中删除，代码对于 spark 运行良好，但不适用于 JNA。因此显然向 中Structure添加了许多字段MyStruct，其中一个不受支持/无法映射。因为无论如何我都不需要这些字段，我如何制定我的代码，以便只MyStruct编码“我的”字段，而不是那些继承自的字段Structure？

标签： javaapache-sparkjna