将 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

将其转换为 RDD，然后将其映射到您的 类。这将 return 这些对象的一个​​ rdd：

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

Answer 2

如果您只能访问 DataFrame 并想将其转换为列表 pojo，则应收集数据框并迭代 org.apache.spark.sql.Row 列表以填充 pojo 列表。

或

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

代码：

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