将RDD转换为DataFrame

将RDD转换为DataFrame

一、概述

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

二、使用反射方式将RDD转换为DataFrame

1、说明

Java版本:Spark SQL是支持将包含了JavaBean的RDD转换为DataFrame的。JavaBean的信息,就定义了元数据。
Spark SQL现在是不支持将包含了嵌套JavaBean或者List等复杂数据的JavaBean,作为元数据的。只支持一个包含简单数据类型的field的JavaBean。 Scala版本:而Scala由于其具有隐式转换的特性,所以Spark SQL的Scala接口,是支持自动将包含了case class的RDD转换为DataFrame的。case class就定义了元数据。
Spark SQL会通过反射读取传递给case class的参数的名称,然后将其作为列名。与Java不同的是,Spark SQL是支持将包含了嵌套数据结构的case class作为元数据的,比如包含了Array等。

2、java实现

#students.txt
1,leo,17
2,marry,17
3,jack,18
4,tom,19 -----------------
package cn.spark.study.sql; import java.util.List; import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext; /**
* 使用反射方式将RDD转换为DataFrame
* @author ming
*
*/ public class RDD2DataFrameReflection {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameReflection");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lines = sc.textFile("D:\\test-file\\sql\\students.txt");
JavaRDD<Student> students = lines.map(new Function<String, Student>() { private static final long serialVersionUID = 1L; @Override
public Student call(String line) throws Exception {
String[] lineSplited = line.split(",");
Student stu = new Student();
stu.setId(Integer.valueOf(lineSplited[0].trim()));
stu.setName(lineSplited[1]);
stu.setAge(Integer.valueOf(lineSplited[2].trim())); return stu;
} }); // 使用反射方式,将RDD转换为DataFrame
// 将Student.class传入进去,其实就是用反射的方式来创建DataFrame
// 因为Student.class本身就是反射的一个应用
// 然后底层还得通过对Student Class进行反射,来获取其中的field
// 这里要求,JavaBean必须实现Serializable接口,是可序列化的
DataFrame studentDF = sqlContext.createDataFrame(students, Student.class); //拿到了一个DataFrame之后,就可以将其注册为一个临时表,然后针对其中的数据执行SQL语句
studentDF.registerTempTable("students"); //针对students临时表执行sql语句,查询年龄小于等于18岁的学生
DataFrame teenagerDF = sqlContext.sql("select * from students where age <= 18"); //将查询出来的DataFrame,再次转换为RDD
JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD(); //将RDD中的数据,进行映射,映射为student
JavaRDD<Student> teenagerStudentRDD = teenagerRDD.map(new Function<Row, Student>() { private static final long serialVersionUID = 1L; @Override
public Student call(Row row) throws Exception {
Student stu = new Student();
// row中的数据的顺序,可能是跟我们期望的是不一样的!
stu.setAge(row.getInt(0));
stu.setId(row.getInt(1));
stu.setName(row.getString(2));
return stu;
}
}); //将数据col1ect回来,打印出来
List<Student> studentList = teenagerStudentRDD.collect(); for(Student stu : studentList) {
System.out.println(stu);
} } }

Student类,如下:

package cn.spark.study.sql;

import java.io.Serializable;

public class Student implements Serializable{
private int id;
private String name;
private int age; public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
} @Override
public String toString() {
return "Student [id=" + id + ", name=" + name + ", age=" + age + "]";
} }

3、scala实现

package cn.spark.study.sql

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext /**
* 如果要用scala开发spark程序
* 然后在其中,还要实现基于反射的RDD到DataFrame的转换,就必须得用object extends App的方式
* 不能用def main()方法的方式,来运行程序,否则就会报no typetag for ...class的错误
* @author Administrator
*/
object RDD2DataFrameReflection extends App { val conf = new SparkConf()
.setMaster("local")
.setAppName("RDD2DataFrameReflection")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc) // 在Scala中使用反射方式,进行RDD到DataFrame的转换,需要手动导入一个隐式转换
import sqlContext.implicits._ case class Student(id: Int, name: String, age: Int) // 这里其实就是一个普通的,元素为case class的RDD
// 直接对它使用toDF()方法,即可转换为DataFrame
val studentDF = sc.textFile("D:\\test-file\\sql\\students.txt", 1)
.map { line => line.split(",") }
.map { arr => Student(arr(0).trim().toInt, arr(1), arr(2).trim().toInt) }
.toDF() // 前面讲的DataFrame的常用操作依然适用
studentDF.show()
studentDF.schema
studentDF.select(studentDF.col("id")).show()
studentDF.select(studentDF.col("age")).show()
studentDF.select(studentDF.col("name")).show()
studentDF.select(studentDF.col("id"), studentDF.col("name"), studentDF.col("age").plus(1))
studentDF.filter(studentDF.col("age").gt(18)).show()
studentDF.groupBy(studentDF.col("age")).count.show() studentDF.registerTempTable("students") val teenagerDF = sqlContext.sql("select * from students where age<=18") val teenagerRDD = teenagerDF.rdd // 在scala中,row中的数据的顺序,反而是按照我们期望的来排列的,这个跟java是不一样的哦
teenagerRDD.map { row => Student(row(0).toString().toInt, row(1).toString(), row(2).toString().toInt) }
.collect()
.foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) } // 在scala中,对row的使用,比java中的row的使用,更加丰富
// 在scala中,可以用row的getAs()方法,获取指定列名的列
teenagerRDD.map { row => Student(row.getAs[Int]("id"), row.getAs[String]("name"), row.getAs[Int]("age")) }
.collect()
.foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) } // 还可以通过row的getValuesMap()方法,获取指定几列的值,返回的是个map
val studentRDD = teenagerRDD.map { row => {
val map = row.getValuesMap[Any](Array("id", "name", "age"));
Student(map("id").toString().toInt, map("name").toString(), map("age").toString().toInt)
}
}
studentRDD.collect().foreach { stu => println(stu.id + ":" + stu.name + ":" + stu.age) }
}

三、使用编程方式将RDD转换为DataFrame

1、说明

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

2、java实现

package cn.spark.study.sql;

import java.util.ArrayList;
import java.util.List; import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.StructField; /**
* 以编程方式动态指定元数据,将RDD转换为DataFrame
* @author Administrator
*
*/
public class RDD2DataFrameProgrammatically { public static void main(String[] args) {
// 创建SparkConf、JavaSparkContext、SQLContext
SparkConf conf = new SparkConf()
.setMaster("local")
.setAppName("RDD2DataFrameProgrammatically");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc); // 第一步,创建一个普通的RDD,但是,必须将其转换为RDD<Row>的这种格式
JavaRDD<String> lines = sc.textFile("D:\\test-file\\sql\\students.txt"); // 分析一下
// 它报了一个,不能直接从String转换为Integer的一个类型转换的错误
// 就说明什么,说明有个数据,给定义成了String类型,结果使用的时候,要用Integer类型来使用
// 而且,错误报在sql相关的代码中
// 所以,基本可以断定,就是说,在sql中,用到age<=18的语法,所以就强行就将age转换为Integer来使用
// 但是,肯定是之前有些步骤,将age定义为了String
// 所以就往前找,就找到了这里
// 往Row中塞数据的时候,要注意,什么格式的数据,就用什么格式转换一下,再塞进去
JavaRDD<Row> studentRDD = lines.map(new Function<String, Row>() { private static final long serialVersionUID = 1L; @Override
public Row call(String line) throws Exception {
String[] lineSplited = line.split(",");
return RowFactory.create(
Integer.valueOf(lineSplited[0]),
lineSplited[1],
Integer.valueOf(lineSplited[2]));
} }); // 第二步,动态构造元数据
// 比如说,id、name等,field的名称和类型,可能都是在程序运行过程中,动态从mysql db里
// 或者是配置文件中,加载出来的,是不固定的
// 所以特别适合用这种编程的方式,来构造元数据
List<StructField> structFields = new ArrayList<StructField>();
structFields.add(DataTypes.createStructField("id", DataTypes.IntegerType, true));
structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
StructType structType = DataTypes.createStructType(structFields); // 第三步,使用动态构造的元数据,将RDD转换为DataFrame
DataFrame studentDF = sqlContext.createDataFrame(studentRDD, structType); // 后面,就可以使用DataFrame了
studentDF.registerTempTable("students"); DataFrame teenagerDF = sqlContext.sql("select * from students where age<=18"); List<Row> rows = teenagerDF.javaRDD().collect();
for(Row row : rows) {
System.out.println(row);
}
} }

3、scala实现

package cn.spark.study.sql

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructType,StructField,StringType,IntegerType}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext /**
* @author Administrator
*/ object RDD2DataFrameProgrammatically extends App { val conf = new SparkConf()
.setMaster("local")
.setAppName("RDD2DataFrameProgrammatically")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc) // 第一步,构造出元素为Row的普通RDD
val studentRDD = sc.textFile("D:\\test-file\\sql\\students.txt", 1)
.map { line => Row(line.split(",")(0).toInt, line.split(",")(1), line.split(",")(2).toInt) } // 第二步,编程方式动态构造元数据
val structType = StructType(Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("age", IntegerType, true))) // 第三步,进行RDD到DataFrame的转换
val studentDF = sqlContext.createDataFrame(studentRDD, structType) // 继续正常使用
studentDF.registerTempTable("students") val teenagerDF = sqlContext.sql("select * from students where age<=18") val teenagerRDD = teenagerDF.rdd.collect().foreach { row => println(row) }
}
05-11 16:59