Наиболее эффективный способ превратить Спарк SQL DataFrame в список POJOs

Question

Предположим, вы следующий искровой 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-бобов следующую структуру

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 и хотите, чтобы преобразовать его в списке POJOs, вы должны собрать dataframe и перебирать список org.apache.spark.sql.Row для заполнения списка POJOs.

Или

Вы можете использовать разъем искровой Кассандры, который содержит методы для создания 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(); 

Answer 2

Преобразуйте его в RDD, а затем просто переместите его в свои классы. Это вернет РД этих объектов:

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