我們前面說了，DataFrame的數(shù)據(jù)來源可以是RDD，這樣的話，我們就可以直接針對(duì)HDFS等任何可以構(gòu)建為RDD的數(shù)據(jù)，使用SparkSQL進(jìn)行SQL查詢了，這個(gè)功能是無比強(qiáng)大的，比如說，我們可以直接使用SQL查詢HDFS中的數(shù)據(jù)。
SparkSQL支持兩種方式來將RDD轉(zhuǎn)換為DataFrame
-
1、反射方式：使用反射來推斷包含了特定數(shù)據(jù)類型的RDD的元數(shù)據(jù)；這種基于反射的方式，代碼比較簡(jiǎn)潔，當(dāng)你已經(jīng)知道RDD的元數(shù)據(jù)時(shí)，是一種非常不錯(cuò)的方式；
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轉(zhuǎn)換為DataFrame */ public class RDD2DataFrameReflection { public static void main(String[] args) { // 創(chuàng)建普通的RDD SparkConf conf = new SparkConf() .setMaster("local") .setAppName("RDD2DataFrameReflection"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lines =sc.textFile("C://Users//Administrator//Desktop//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轉(zhuǎn)換為DataFrame // 將Student.class傳入進(jìn)去，其實(shí)就是用反射的方式來創(chuàng)建DataFrame // 因?yàn)镾tudent.class本身就是反射的一個(gè)應(yīng)用 // 然后底層還得通過對(duì)Student Class進(jìn)行反射，來獲取其中的field // 這里要求，JavaBean必須實(shí)現(xiàn)Serializable接口，是可序列化的 DataFrame studentDF = sqlContext.createDataFrame(students, Student.class); // 拿到了一個(gè)DataFrame之后， //就可以將其注冊(cè)為一個(gè)臨時(shí)表，然后針對(duì)其中的數(shù)據(jù)執(zhí)行SQL語句 studentDF.registerTempTable("students"); // 針對(duì)students臨時(shí)表執(zhí)行SQL語句， //查詢年齡小于等于18歲的學(xué)生，就是teenageer DataFrame teenagerDF = sqlContext.sql("select * from students where age<= 18"); // 將查詢出來的DataFrame，再次轉(zhuǎn)換為RDD JavaRDD<Row> teenagerRDD = teenagerDF.javaRDD(); // 將RDD中的數(shù)據(jù)，進(jìn)行映射，映射為Student JavaRDD<Student> teenagerStudentRDD = teenagerRDD.map(new Function<Row, Student>() { private static final long serialVersionUID = 1L; @Override public Student call(Row row) throws Exception { // row中的數(shù)據(jù)的順序，可能是跟我們期望的是不一樣的！ Student stu = new Student(); stu.setAge(row.getInt(0)); stu.setId(row.getInt(1)); stu.setName(row.getString(2)); return stu; } }); // 將數(shù)據(jù)collect回來，打印出來 List<Student> studentList = teenagerStudentRDD.collect(); for(Student stu : studentList) { System.out.println(stu); } } }scala版本：
package cn.spark.study.sql import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.sql.SQLContext /** * 如果要用scala開發(fā)spark程序，然后在其中， * 還要實(shí)現(xiàn)基于反射的RDD到DataFrame的轉(zhuǎn)換， * 就必須得用object extends App的方式，不能用def main() * 方法的方式來運(yùn)行程序，否則就會(huì)報(bào)no typetag for ...class的錯(cuò)誤 */ object RDD2DataFrameReflection extends App { val conf = new SparkConf() .setMaster("local") .setAppName("RDD2DataFrameReflection") val sc = new SparkContext(conf) val sqlContext = new SQLContext(sc) // 在Scala中使用反射方式，進(jìn)行RDD到DataFrame的轉(zhuǎn)換，需要手動(dòng)導(dǎo)入一個(gè)隱式轉(zhuǎn)換 import sqlContext.implicits._ case class Student(id: Int, name: String, age: Int) // 這里其實(shí)就是一個(gè)普通的，元素為case class的RDD // 直接對(duì)它使用toDF()方法，即可轉(zhuǎn)換為DataFrame val studentDF = sc.textFile("C://Users//Administrator//Desktop//students.txt", 1) .map { line => line.split(",") } .map { arr => Student(arr(0).trim().toInt, arr(1), arr(2).trim().toInt) } .toDF() studentDF.registerTempTable("students") val teenagerDF = sqlContext.sql("select * from students where age<=18") val teenagerRDD = teenagerDF.rdd // 在scala中，row中的數(shù)據(jù)的順序，反而是按照我們期望的來排列的，這個(gè)跟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中，對(duì)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()方法，獲取指定幾列的值，返回的是個(gè)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) } } -
2、編程接口：通過編程接口來創(chuàng)建DataFrame，可以在程序運(yùn)行時(shí)動(dòng)態(tài)構(gòu)建一份元數(shù)據(jù)，然后將其應(yīng)用到已經(jīng)存在的RDD上。這種方式的代碼比較冗長(zhǎng)，但是如果在編寫程序時(shí)，還不知道RDD的元數(shù)據(jù)，只有在程序運(yùn)行時(shí)，才能動(dòng)態(tài)得知其元數(shù)據(jù)，那么只能通過這種動(dòng)態(tài)構(gòu)建元數(shù)據(jù)的方式實(shí)現(xiàn)了；
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; /** * 以編程方式動(dòng)態(tài)指定元數(shù)據(jù)，將RDD轉(zhuǎn)換為DataFrame */ public class RDD2DataFrameProgrammatically { public static void main(String[] args) { // 創(chuàng)建SparkConf、JavaSparkContext、SQLContext SparkConf conf = new SparkConf() .setMaster("local") .setAppName("RDD2DataFrameProgrammatically"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); // 第一步，創(chuàng)建一個(gè)普通的RDD，但是，必須將其轉(zhuǎn)換為RDD<Row>的這種格式 JavaRDD<String> lines = sc.textFile("C://Users//Administrator//Desktop//students.txt"); // 往Row中塞數(shù)據(jù)的時(shí)候，要注意，什么格式的數(shù)據(jù)，就用什么格式轉(zhuǎn)換一下，再塞進(jìn)去 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])); } }); // 第二步，動(dòng)態(tài)構(gòu)造元數(shù)據(jù) // 比如說，id、name等，field的名稱和類型，可能都是在程序運(yùn)行過程中，動(dòng)態(tài)從mysql db里 // 或者是配置文件中，加載出來的，是不固定的 // 所以特別適合用這種編程的方式，來構(gòu)造元數(shù)據(jù) 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); // 第三步，使用動(dòng)態(tài)構(gòu)造的元數(shù)據(jù)，將RDD轉(zhuǎn)換為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); } } }
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
object RDD2DataFrameProgrammatically extends App {
val conf = new SparkConf()
.setMaster("local")
.setAppName("RDD2DataFrameProgrammatically")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
// 第一步，構(gòu)造出元素為Row的普通RDD
val studentRDD = sc.textFile("C://Users//Administrator//Desktop//students.txt", 1)
.map { line => Row(line.split(",")(0).toInt, line.split(",")(1), line.split(",")(2).toInt) }
// 第二步，編程方式動(dòng)態(tài)構(gòu)造元數(shù)據(jù)
val structType = StructType(Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("age", IntegerType, true)))
// 第三步，進(jìn)行RDD到DataFrame的轉(zhuǎn)換
val studentDF = sqlContext.createDataFrame(studentRDD, structType)
// 繼續(xù)正常使用
studentDF.registerTempTable("students")
val teenagerDF = sqlContext.sql("select * from students where age<=18")
val teenagerRDD = teenagerDF.rdd.collect().foreach { row => println(row) }
}
