第59课：使用Java和Scala在IDE中实战RDD和DataFrame-经管之家官网！

坛友互助群

扫码加入各岗位、行业、专业交流群

一、Java方式：

1. public class RDD2DataFrameByReflection {
   
2. public static void main(String args[]){
   
3. SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameByReflection");
   
4. JavaSparkContext sc = new JavaSparkContext(conf);
   
5. SQLContext sqlContext = new SQLContext(sc);
   
6. 
   
7. JavaRDD<String> lines = sc.textFile("D:\\idea\\spark_dt\\src\\main\\java\\com\\base\\spark\\sql\\transfer\\person.txt");
   
8. 
   
9. JavaRDD<Person_> persons = lines.map((Function<String, Person_>) line -> {
   
10. String[] splited = line.split(",");
    
11. Person_ p = new Person_();
    
12. p.setId(Integer.valueOf(splited[0]));
    
13. p.setName(splited[1]);
    
14. p.setAge(Integer.valueOf(splited[0]));
    
15. return p;
    
16. });
    
17. //在底层通过反射的方式获得P鹅绒的所有的fields,结合RDD本身，就生成了DataFrame
    
18. DataFrame df = sqlContext.createDataFrame(persons, Person_.class);
    
19. df.show();
    
20. df.registerTempTable("persons");
    
21. DataFrame bigDatas = sqlContext.sql("select * from persons where age >= 6");
    
22. JavaRDD<Row> bigDataRDD= bigDatas.toJavaRDD();
    
23. JavaRDD<Person_> personRDD = bigDataRDD.map((Function<Row, Person_>) row -> {
    
24. Person_ p = new Person_();
    
25. p.setId(row.getInt(0));
    
26. p.setName(row.getString(1));
    
27. p.setAge(row.getInt(2));
    
28. return p;
    
29. });
    
30. personRDD.collect().forEach(System.out::println);
    
31. 
    
32. }
    
33. }
    
34. 
    
35. class Person_{
    
36. public int getId() {
    
37. return id;
    
38. }
    
39. 
    
40. public void setId(Integer id) {
    
41. this.id = id;
    
42. }
    
43. 
    
44. public String getName() {
    
45. return name;
    
46. }
    
47. 
    
48. public void setName(String name) {
    
49. this.name = name;
    
50. }
    
51. 
    
52. public int getAge() {
    
53. return age;
    
54. }
    
55. 
    
56. public void setAge(Integer age) {
    
57. this.age = age;
    
58. }
    
59. 
    
60. @Override
    
61. public String toString() {
    
62. return "Person{" +
    
63. "id=" + id +
    
64. ", name='" + name + '\'' +
    
65. ", age=" + age +
    
66. '}';
    
67. }
    
68. 
    
69. public Person_(Integer id, String name, Integer age) {
    
70. this.id = id;
    
71. this.name = name;
    
72. this.age = age;
    
73. }
    
74. 
    
75. Person_(){}
    
76. private Integer id;
    
77. private String name;
    
78. private Integer age;
    
79. }

运行结果如下：

1. 16/03/25 05:59:14 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, localhost): java.lang.IllegalAccessException: Class org.apache.spark.sql.SQLContext$anonfun$org$apache$spark$sql$SQLContext$beansToRows$1$anonfun$apply$1 can not access a member of class com.base.spark.sql.transfer.Person_ with modifiers "public"
   
2. at sun.reflect.Reflection.ensureMemberAccess(Reflection.java:102)

解决办法：Person_类需要Public修饰符，将Person_放到单独的文件

运行结果如下：

1. 16/03/25 06:04:46 INFO DAGScheduler: Job 0 finished: show at RDD2DataFrameByReflection.java:32, took 0.178583 s
   
2. +---+---+------+
   
3. |age| id|name|
   
4. +---+---+------+
   
5. |1|1| spark|
   
6. |2|2|hadoop|
   
7. |3|3| flink|
   
8. |4|4| spark|
   
9. |5|5|hadoop|
   
10. |6|6| spark|
    
11. |7|7| spark|
    
12. |8|8| spark|
    
13. |9|9| scala|
    
14. | 10| 10|java|
    
15. | 11| 11| spark|
    
16. +---+---+------+
    
17. 
    
18. 16/03/25 06:04:47 INFO SparkContext: Starting job: collect at RDD2DataFrameByReflection.java:43
    
19. 16/03/25 06:04:47 INFO DAGScheduler: Got job 1 (collect at RDD2DataFrameByReflection.java:43) with 1 output partitions
    
20. 16/03/25 06:04:47 INFO DAGScheduler: Final stage: ResultStage 1 (collect at RDD2DataFrameByReflection.java:43)
    
21. 16/03/25 06:04:47 INFO DAGScheduler: Parents of final stage: List()
    
22. 16/03/25 06:04:47 INFO DAGScheduler: Missing parents: List()
    
23. 16/03/25 06:04:47 INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[8] at map at RDD2DataFrameByReflection.java:36), which has no missing parents
    
24. 16/03/25 06:04:47 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 9.5 KB, free 135.6 KB)
    
25. 16/03/25 06:04:47 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 4.9 KB, free 140.5 KB)
    
26. 16/03/25 06:04:47 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory on localhost:56952 (size: 4.9 KB, free: 1771.1 MB)
    
27. 16/03/25 06:04:47 INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:1006
    
28. 16/03/25 06:04:47 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[8] at map at RDD2DataFrameByReflection.java:36)
    
29. 16/03/25 06:04:47 INFO TaskSchedulerImpl: Adding task set 1.0 with 1 tasks
    
30. 16/03/25 06:04:47 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1, localhost, partition 0,PROCESS_LOCAL, 2177 bytes)
    
31. 16/03/25 06:04:47 INFO Executor: Running task 0.0 in stage 1.0 (TID 1)
    
32. 16/03/25 06:04:47 INFO HadoopRDD: Input split: file:/D:/idea/spark_dt/src/main/java/com/base/spark/sql/transfer/person.txt:0+131
    
33. 16/03/25 06:04:47 INFO BlockManagerInfo: Removed broadcast_1_piece0 on localhost:56952 in memory (size: 3.1 KB, free: 1771.1 MB)
    
34. 16/03/25 06:04:47 INFO ContextCleaner: Cleaned accumulator 1
    
35. 16/03/25 06:04:47 INFO GeneratePredicate: Code generated in 240.433533 ms
    
36. 16/03/25 06:04:47 INFO GenerateUnsafeProjection: Code generated in 13.091797 ms
    
37. 16/03/25 06:04:47 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
    
38. java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String

错误原因：返回的ROW中的字段并没有按照Person_字段的顺序排列

Row中的字段顺序：

可以看到这里scheme的顺序并不是我们定义的字段顺序

解决办法：

1. /**
   
2. * Returns the value of a given fieldName.
   
3. * For primitive types if value is null it returns 'zero value' specific for primitive
   
4. * ie. 0 for Int - use isNullAt to ensure that value is not null
   
5. *
   
6. * @throws UnsupportedOperationException when schema is not defined.
   
7. * @throws IllegalArgumentException when fieldName do not exist.
   
8. * @throws ClassCastException when data type does not match.
   
9. */
   
10. def getAs[T](fieldName: String): T = getAs[T](fieldIndex(fieldName))

可以看到这里Row 提供了根据Row中的字段名称来获取字段的值

修改代码：

1. JavaRDD<Person_> personRDD = bigDataRDD.map((Function<Row, Person_>) row -> {
   
2. Person_ p = new Person_();
   
3. p.setId(row.getAs("id"));
   
4. p.setName(row.getAs("name"));
   
5. p.setAge(row.getAs("age"));
   
6. return p;
   
7. });

运行结果：

1. java.io.NotSerializableException: com.base.spark.sql.transfer.Person_
   
2. Serialization stack:
   
3. - object not serializable (class: com.base.spark.sql.transfer.Person_, value: Person{id=6, name='spark', age=6})
   
4. - element of array (index: 0)
   
5. - array (class [Ljava.lang.Object;, size 6)
   
6. at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
   
7. at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
   
8. at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:101)
   
9. at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:239)
   
10. at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    
11. at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    
12. at java.lang.Thread.run(Thread.java:745)

提示我们Person_ 没有序列化

解决办法：

让Person_ 实现Serializable 接口

再次运行：

1. 16/03/25 06:18:00 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool
   
2. 16/03/25 06:18:00 INFO DAGScheduler: ResultStage 1 (collect at RDD2DataFrameByReflection.java:43) finished in 0.347 s
   
3. 16/03/25 06:18:00 INFO DAGScheduler: Job 1 finished: collect at RDD2DataFrameByReflection.java:43, took 0.356079 s
   
4. Person{id=6, name='spark', age=6}
   
5. Person{id=7, name='spark', age=7}
   
6. Person{id=8, name='spark', age=8}
   
7. Person{id=9, name='scala', age=9}
   
8. Person{id=10, name='java', age=10}
   
9. Person{id=11, name='spark', age=11}

这次运行成功了

总结：

JavaRDD和DataFrame互相转换注意事项：

1）反射的类必须是Pulicclass

2）定义的类必须实现Serializable 接口

3）DataFrame 转换成RDD时注意根据fieldName 获取对应的值

二、scala方式

注意：这里lines.map 返回的RDD并没有.toDF 方法，这里用到了隐式转换

1. /**
   
2. * Creates a DataFrame from an RDD of Product (e.g. case classes, tuples).
   
3. * @since 1.3.0
   
4. */
   
5. implicit def rddToDataFrameHolder[A <: Product : TypeTag](rdd: RDD[A]): DataFrameHolder = {
   
6. DataFrameHolder(_sqlContext.createDataFrame(rdd))
   
7. }

隐式转换最终将RDD转换成了DataFrameHolder 然后调用toDF 最终将RDD转换成了DataFrame了

1. case class DataFrameHolder private[sql](private val df: DataFrame) {
   
2. 
   
3. // This is declared with parentheses to prevent the Scala compiler from treating
   
4. // `rdd.toDF("1")` as invoking this toDF and then apply on the returned DataFrame.
   
5. def toDF(): DataFrame = df
   
6. 
   
7. def toDF(colNames: String*): DataFrame = df.toDF(colNames : _*)
   
8. }

注：本学习笔记来自DT大数据梦工厂 微信公众号：DT_Spark 每晚8点YY永久直播频道：68917580

扫码或添加微信号：坛友素质互助

「经管之家」APP：经管人学习、答疑、交友，就上经管之家！
免流量费下载资料----在经管之家app可以下载论坛上的所有资源，并且不额外收取下载高峰期的论坛币。
涵盖所有经管领域的优秀内容----覆盖经济、管理、金融投资、计量统计、数据分析、国贸、财会等专业的学习宝库，各类资料应有尽有。
来自五湖四海的经管达人----已经有上千万的经管人来到这里，你可以找到任何学科方向、有共同话题的朋友。
经管之家（原人大经济论坛），跨越高校的围墙，带你走进经管知识的新世界。
扫描下方二维码下载并注册APP