RDD转换为DataFrame

  1. 为什么要将RDD转换为DataFrame？因为这样的话，我们就可以直接针对HDFS等任何可以构建为RDD的数据，使用Spark SQL进行SQL查询了。这个功能是无比强大的。想象一下，针对HDFS中的数据，直接就可以使用SQL进行查询。
  3. Spark SQL支持两种方式来将RDD转换为DataFrame。
  5. 第一种方式，是使用反射来推断包含了特定数据类型的RDD的元数据。这种基于反射的方式，代码比较简洁，当你已经知道你的RDD的元数据时，是一种非常不错的方式。
  7. 第二种方式，是通过编程接口来创建DataFrame，你可以在程序运行时动态构建一份元数据，然后将其应用到已经存在的RDD上。这种方式的代码比较冗长，但是如果在编写程序时，还不知道RDD的元数据，只有在程序运行时，才能动态得知其元数据，那么只能通过这种动态构建元数据的方式。

RDD转换为DataFrame

  1. 为什么要将RDD转换为DataFrame？因为这样的话，我们就可以直接针对HDFS等任何可以构建为RDD的数据，使用Spark SQL进行SQL查询了。这个功能是无比强大的。想象一下，针对HDFS中的数据，直接就可以使用SQL进行查询。
  3. Spark SQL支持两种方式来将RDD转换为DataFrame。
  5. 第一种方式，是使用反射来推断包含了特定数据类型的RDD的元数据。这种基于反射的方式，代码比较简洁，当你已经知道你的RDD的元数据时，是一种非常不错的方式。
  7. 第二种方式，是通过编程接口来创建DataFrame，你可以在程序运行时动态构建一份元数据，然后将其应用到已经存在的RDD上。这种方式的代码比较冗长，但是如果在编写程序时，还不知道RDD的元数据，只有在程序运行时，才能动态得知其元数据，那么只能通过这种动态构建元数据的方式。

使用编程方式指定元数据

  1. Java版本：当JavaBean无法预先定义和知道的时候，比如要动态从一个文件中读取数据结构，那么就只能用编程方式动态指定元数据了。首先要从原始RDD创建一个元素为Row的RDD；其次要创建一个StructType，来代表Row；最后将动态定义的元数据应用到RDD<Row>上。
  3. Scala版本：Scala的实现方式，与Java是基本一样的。

利用反射方式
java版

  1. /**
  2.  * 使用反射的方式将RDD转换为DataFrame
  3.  * @author Administrator
  4.  *
  5.  */
  6. public class RDD2DataFrameReflection {
  8.     public static void main(String[] args) {
  9.         // 创建普通的RDD
  10.         SparkConf conf = new SparkConf()
  11.                 .setMaster("local")  
  12.                 .setAppName("RDD2DataFrameReflection");  
  13.         JavaSparkContext sc = new JavaSparkContext(conf);
  14.         SQLContext sqlContext = new SQLContext(sc);
  16.         JavaRDD<String> lines = sc.textFile("C://Users//Administrator//Desktop//students.txt");
  18.         JavaRDD<Student> students = lines.map(new Function<String, Student>() {
  20.             private static final long serialVersionUID = 1L;
  22.             @Override
  23.             public Student call(String line) throws Exception {
  24.                 String[] lineSplited = line.split(",");  
  25.                 Student stu = new Student();
  26.                 stu.setId(Integer.valueOf(lineSplited[0].trim()));  
  27.                 stu.setName(lineSplited[1]);  
  28.                 stu.setAge(Integer.valueOf(lineSplited[2].trim())); 
  29.                 return stu;
  30.             }
  32.         });
  34.         // 使用反射方式，将RDD转换为DataFrame
  35.         // 将Student.class传入进去，其实就是用反射的方式来创建DataFrame
  36.         // 因为Student.class本身就是反射的一个应用
  37.         // 然后底层还得通过对Student Class进行反射，来获取其中的field
  38.         // 这里要求，JavaBean必须实现Serializable接口，是可序列化的
  39.         DataFrame studentDF = sqlContext.createDataFrame(students, Student.class);
  41.         // 拿到了一个DataFrame之后，就可以将其注册为一个临时表，然后针对其中的数据执行SQL语句
  42.         studentDF.registerTempTable("students");  
  44.         // 针对students临时表执行SQL语句，查询年龄小于等于18岁的学生，就是teenageer
  45.         DataFrame teenagerDF = sqlContext.sql("select * from students where age<= 18");  
  47.         // 将查询出来的DataFrame，再次转换为RDD
  48.         JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD();
  50.         // 将RDD中的数据，进行映射，映射为Student
  51.         JavaRDD<Student> teenagerStudentRDD = teenagerRDD.map(new Function<Row, Student>() {
  53.             private static final long serialVersionUID = 1L;
  55.             @Override
  56.             public Student call(Row row) throws Exception {
  57.                 // row中的数据的顺序，可能是跟我们期望的是不一样的！
  58.                 Student stu = new Student();
  59.                 stu.setAge(row.getInt(0));
  60.                 stu.setId(row.getInt(1));
  61.                 stu.setName(row.getString(2));
  62.                 return stu;
  63.             }
  65.         });
  67.         // 将数据collect回来，打印出来
  68.         List<Student> studentList = teenagerStudentRDD.collect();
  69.         for(Student stu : studentList) {
  70.             System.out.println(stu);  
  71.         }
  72.     }
  74. }

scala 版

  1. /**
  2.  * 如果要用scala开发spark程序
  3.  * 然后在其中，还要实现基于反射的RDD到DataFrame的转换，就必须得用object extends App的方式
  4.  * 不能用def main()方法的方式，来运行程序，否则就会报no typetag for ...class的错误
  5.  * @author Administrator
  6.  */
  7. object RDD2DataFrameReflection extends App {
  9.   val conf = new SparkConf()
  10.       .setMaster("local")  
  11.       .setAppName("RDD2DataFrameReflection")  
  12.   val sc = new SparkContext(conf)
  13.   val sqlContext = new SQLContext(sc)
  15.   // 在Scala中使用反射方式，进行RDD到DataFrame的转换，需要手动导入一个隐式转换
  16.   import sqlContext.implicits._  
  18.   case class Student(id: Int, name: String, age: Int)
  20.   // 这里其实就是一个普通的，元素为case class的RDD
  21.   // 直接对它使用toDF()方法，即可转换为DataFrame
  22.   val studentDF = sc.textFile("C://Users//Administrator//Desktop//students.txt", 1)
  23.       .map { line => line.split(",") }
  24.       .map { arr => Student(arr(0).trim().toInt, arr(1), arr(2).trim().toInt) }
  25.       .toDF()    
  27.   studentDF.registerTempTable("students")  
  29.   val teenagerDF = sqlContext.sql("select * from students where age<=18")  
  31.   val teenagerRDD = teenagerDF.rdd
  33.   // 在scala中，row中的数据的顺序，反而是按照我们期望的来排列的，这个跟java是不一样的哦
  34.   teenagerRDD.map { row => Student(row(0).toString().toInt, row(1).toString(), row(2).toString().toInt) }
  35.       .collect()
  36.       .foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) }  
  38.   // 在scala中，对row的使用，比java中的row的使用，更加丰富
  39.   // 在scala中，可以用row的getAs()方法，获取指定列名的列
  40.   teenagerRDD.map { row => Student(row.getAs[Int]("id"), row.getAs[String]("name"), row.getAs[Int]("age")) }
  41.       .collect()
  42.       .foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) }
  44.   // 还可以通过row的getValuesMap()方法，获取指定几列的值，返回的是个map
  45.   val studentRDD = teenagerRDD.map { row => {
  46.       val map = row.getValuesMap[Any](Array("id", "name", "age")); 
  47.       Student(map("id").toString().toInt, map("name").toString(), map("age").toString().toInt)    
  48.     } 
  49.   }  
  50.   studentRDD.collect().foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) }  
  51. }