如何將Spark DataFrame轉換爲Java中POJO的RDD

Question

我對Spark很新。我想將DataFrame轉換爲POJO的RDD。像：

JavaRDD<POJOClass> data = df.toJavaRDD(); 

其中df是一個數據框。

df.show()給出：

+---------+---------+---------+---------+          
| A | B | C | D | 
+---------+---------+---------+---------+ 
|603300042|  1025|  2|127000948| 
|603303766|  1112|  2|127000364| 
|603302691|  1184|  2|127000853| 
|603303766|  1112|  2|127000364| 
|603302691|  1184|  2|127000853| 
|603303766|  1112|  2|127000364| 
|603303787|  1041|  2|137000323| 
|603306351|  1041|  2|137000468| 
|603304009|  1307|  2|137000788| 
|603303830|  1041|  2|137000012| 
|603301119|  1002|  2|137000369| 
|603301507|  1188|  2|137001568| 
|603302168|  1041|  2|137000468| 
+---------+---------+---------+---------+ 

我的POJO類如下：

public static class POJOClass { 
     public Long A; 
     public Integer B; 
     public Integer C; 
     public Long D; 
} 

我知道

JavaRDD<Row> data = df.toJavaRDD(); 

效果很好。但是，我有什麼解決辦法可以實現？

Answer 1

可以使用的數據集

public static class POJOClass implements serializable{ 
     public Long A; 
     public Integer B; 
     public Integer C; 
     public Long D; 
} 

    Dataset<POJOClass> pojos = context.read().json("/data.json").as(Encoders.bean(POJOClass.class)); 

Answer 2

試試這個（未測試）：

JavaPairRDD<Long, POJOClass> jpRDD = jdbcDF.toJavaRDD().mapToPair(new PairFunction<Row, Long, POJOClass>() { 

     public Tuple2<Long, POJOClass> call(Row row) throws Exception { 
      POJOClass yourPojo = new POJOClass(); 
      // Fill your pojo using row.get(index) 
      return new Tuple2<Long, POJOClass>(anIndex , (POJOClass) yourPojo); 

     } 

    }); 

Answer 3

您可以使用地圖功能，如下圖所示。

import org.apache.spark.api.java.function.Function; 

JavaRDD<POJOClass> data = df.toJavaRDD().map(new Function<Row, POJOClass>() { 
     @Override 
     public POJOClass call(Row row) { 
     POJOClass pojo = new POJOClass(); 
     pojo.setA(row.getLong(0)); 
     pojo.setB(row.getInt(1)); 
     pojo.setC(row.getInt(2)); 
     pojo.setD(row.getLong(3)); 
     return pojo; 
    } 
});