将Spark SQL DataFrame转换为pojo列表的最有效方法

Question

假设您从Cassandra中提取了以下Spark DataFrame：

DataFrame df = cassandraSqlContext.sql(query);

具有以下

+-----------------+------+-----------------+-----------------------------------------------------+
|assetid          |tslice|deviceid         |value                                                |
+-----------------+------+-----------------+-----------------------------------------------------+
|085eb9c6-8a16-...|201509|085eb9c6-8a16-...|Map(xval -> 120000, type -> xsd:double, yval -> 53.0)|
|085eb9c6-8a16-...|201509|085eb9c6-8a16-...|Map(xval -> 120000, type -> xsd:double, yval -> 53.0)|
|085eb9c6-8a16-...|201509|085eb9c6-8a16-...|Map(xval -> 120000, type -> xsd:double, yval -> 53.0)|
    ...

我想把这个DataFrame转换成一个Java bean列表，其结构如下

public class DataItem {
    private UUID assetID;
    private int tslice;
    private UUID deviceID;
    private Value value;
    // getters, setters...
}

和

public class Value {
    private double xval;
    private String type;
    private double yval;
    // getters, setters...
}

在Spark中，无论从性能还是简洁性来看，最好的方法是什么？

谢谢！

Answer 1

如果您刚刚可以访问DataFrame并希望将其转换为列表pojo，则应收集数据帧并迭代org.apache.spark.sql.Row列表以填充pojo列表。

或

您可以使用spark-cassandra连接器，该连接器包含创建JavaRDD的方法，可以收集JavaRDD以获得pojo列表。

代码：

SparkContextJavaFunctions functions = CassandraJavaUtil.javaFunctions(sparkContext);
JavaRDD<DataItem> cassandraRowsRDD = functions.cassandraTable("keyspace", "table_name",
                           CassandraJavaUtil.mapRowTo(DataItem.class));
//required list of pojos
List<DataItem> = cassandraRowsRDD.collect();

Answer 2

将其转换为RDD，然后将其映射到您的类中。这将返回这些对象的rdd：

val dataItem = df.rdd.map(line => DataItem(line[0], line[1].toInt ...)
val value = df.rdd.map(line => Value(line[10].toDouble, ... )