如何使用 Spark Cassandra 连接器保存 Java bean？

问题描述

我已经阅读了 Spark 文档，但不确定如何使用 Spark Cassandra 连接器将 Java bean 保存到表中？

public class NewImageMetadaTarow implements Serializable {

    private final String merchant;
    private final String productId;
    private final String url;
    private final int width;
    private final int height;

    public NewImageMetadaTarow(NewImageMetadaTarow row) {
        this.merchant = row.getMerchant();
        this.productId = row.getProductId();
        this.url = row.getUrl();
        this.width = row.getWidth();
        this.height = row.getHeight();
    }

    public String getMerchant() {
        return merchant;
    }

    public String getProductId() {
        return productId;
    }

    public String getUrl() {
        return url;
    }

    public int getWidth() {
        return width;
    }

    public int getHeight() {
        return height;
    }
}

我有一个 RDD RDD[NewImageMetadaTarow]，我想像这样保存：

myRDD.savetoCassandra(keyspace,"imageMetadatav3",SomeColumns("merchant","productid","url"))

这会导致此错误：

java.lang.IllegalArgumentException: requirement Failed: Columns not found in com.mridang.image.NewImageMetadaTarow: [merchant,productid,url]
    at scala.Predef$.require(Predef.scala:281)
    at com.datastax.spark.connector.mapper.DefaultColumnMapper.columnMapForWriting(DefaultColumnMapper.scala:106)
    at com.datastax.spark.connector.mapper.MappedToGettableDataConverter$$anon$1.<init>(MappedToGettableDataConverter.scala:35)
    at com.datastax.spark.connector.mapper.MappedToGettableDataConverter$.apply(MappedToGettableDataConverter.scala:26)
    at com.datastax.spark.connector.writer.DefaultRowWriter.<init>(DefaultRowWriter.scala:16)
    at com.datastax.spark.connector.writer.DefaultRowWriter$$anon$1.rowWriter(DefaultRowWriter.scala:30)
    at com.datastax.spark.connector.writer.DefaultRowWriter$$anon$1.rowWriter(DefaultRowWriter.scala:28)
    at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:423)
    at com.datastax.spark.connector.RDDFunctions.savetoCassandra(RDDFunctions.scala:35)

根据我的理解（以及糟糕的 Scala foo），它似乎无法从 Java bean 中推断出属性名称。

另一个问题是我表中的列名都是小写的，删除了空格和连字符，即 getter getProductId 的相应 Cassandra 列是 productid。

（如果我使用 Jackson，我可以简单地添加 JsonProperty 注释。我想知道我可以用 Cassadra Mapper 做同样的事情。?）

解决方法

这花了一些时间，但结果是这样的：

val columns: RowWriterFactory[NewImageMetadataRow] =
  CassandraJavaUtil.mapToRow(classOf[NewImageMetadataRow])

myRDD.saveToCassandra(keyspace,"imagemetadatav3")(CassandraConnector(sc),columns)

bean 中的字段需要公开并使用 @CqlName 批注进行批注。

@CqlName("merchant")
public final String merchant;