创建DataFrame
使用SQLContext,可以从RDD、Hive表或者其他数据源,来创建一个DataFrame。以下是一个使用JSON文件创建DataFrame的例子:
Java版本:
JavaSparkContext sc = ...;
SQLContext sqlContext = new SQLContext(sc);
DataFrame df = sqlContext.read().json("hdfs://spark1:9000/students.json");
df.show();
Scala版本:
val sc: SparkContext = ...
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.json("hdfs://spark1:9000/students.json")
df.show()
import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.sql.DataFrame; import org.apache.spark.sql.SQLContext; /** * 使用json文件创建DataFrame * @author swy * */ public class DataFrameCreate { public static void main(String[] args) { SparkConf conf = new SparkConf() .setAppName("DataFrameCreate") .setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); DataFrame df = sqlContext.read().json( "E://swy//resource//workspace-neno//spark-study-java//txt//student.json"); df.show(); } }
二、RDD转换为DataFrame
为什么要将RDD转换为DataFrame?因为这样的话,我们就可以直接针对HDFS等任何可以构建为RDD的数据,使用Spark SQL进行SQL查询了。
Spark SQL支持两种方式来将RDD转换为DataFrame:
第一种方式,是使用反射来推断包含了特定数据类型的RDD的元数据。这种基于反射的方式,代码比较简洁,当你已经知道你的RDD的元数据时,是一种非常不错的方式。
第二种方式,是通过编程接口来创建DataFrame,你可以在程序运行时动态构建一份元数据,然后将其应用到已经存在的RDD上。这种方式的代码比较冗长,但是如果在编写程序时,还不知道RDD的元数据,只有在程序运行时,才能动态得知其元数据,那么只能通过这种动态构建元数据的方式。
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等。
下面源码中students.txt文件如下:
1,lang,17 2,hong,15 3,bai,20 4,zi,18
Java版本:
package swy.study.spark.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; public class RDD2DataFrameProgrammatically { public static void main(String[] args) { SparkConf conf = new SparkConf() .setAppName("DataFrameCreate") .setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lines = sc.textFile( "E://swy//resource//workspace-neno//spark-study-java//txt//students.txt"); JavaRDD<Student> students = lines.map(new Function<String, Student>(){ private static final long serialVersionUID = 1L; public Student call(String line) throws Exception { String[] lineSpited = line.split(","); Student stu = new Student(); stu.setId(Integer.valueOf(lineSpited[0])); stu.setName(lineSpited[1]); stu.setAge(Integer.valueOf(lineSpited[2])); return stu; } }); //使用反射方式,将RDD转换为DataFrame DataFrame studentDF = sqlContext.createDataFrame(students, Student.class); //注册临时表 studentDF.registerTempTable("students"); //针对临时表执行Sql语句(查询年龄小于18岁的学生) //JavaBean必须实现Serializable接口 DataFrame tenagerDF = sqlContext.sql("select * from students where age<=18"); //将查询出来的数据再次转换为RDD JavaRDD<Row> tenagerRDD = tenagerDF.javaRDD(); //将RDD数据映射为Student格式 JavaRDD<Student> tenagerStudents = tenagerRDD.map(new Function<Row, Student>(){ private static final long serialVersionUID = 1L; //顺序和期望的可能不一样 public Student call(Row row) throws Exception { Student stu = new Student(); stu.setId(row.getInt(1)); stu.setName(row.getString(2)); stu.setAge(row.getInt(0)); return stu; } }); //打印 List<Student> studentList = tenagerStudents.collect(); for (Student stu : studentList) { System.out.println("学号: " + stu.getId()); System.out.println("姓名: " + stu.getName()); System.out.println("年龄: " + stu.getAge()); } } }
Scala版本:
package study.spark.scala.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 swy */ 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("E://swy//resource//workspace-neno//spark-study-java//txt//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中的数据的顺序,反而是按照我们期望的来排列的,这个跟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) } }
2、使用编程方式指定元数据
Java版本:当JavaBean无法预先定义和知道的时候,比如要动态从一个文件中读取数据结构,那么就只能用编程方式动态指定元数据了。首先要从原始RDD创建一个元素为Row的RDD;其次要创建一个StructType,来代表Row;最后将动态定义的元数据应用到RDD<Row>上。
Scala版本:Scala的实现方式,与Java基本一样。
Java版本:
package swy.study.spark.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.StructField; import org.apache.spark.sql.types.StructType; /** * 以编程的方式动态指定元数据,将RDD转换为DataFrame * @author swy * */ public class RDD2DataFrameProgrammatically { public static void main(String[] args) { SparkConf conf = new SparkConf() .setAppName("RDD2DataFrameProgrammatically") .setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); JavaRDD<String> lines = sc.textFile( "E://swy//resource//workspace-neno//spark-study-java//txt//students.txt"); JavaRDD<Row> stuRow = lines.map(new Function<String, Row>(){ private static final long serialVersionUID = 1L; public Row call(String line) throws Exception { String[] lineSpited = line.split(","); return RowFactory.create(Integer.valueOf(lineSpited[0]), lineSpited[1], Integer.valueOf(lineSpited[2])); } }); //构造元数据 List<StructField> structFiled = new ArrayList<StructField>(); structFiled.add(DataTypes.createStructField("id", DataTypes.IntegerType, true)); structFiled.add(DataTypes.createStructField("name", DataTypes.StringType, true)); structFiled.add(DataTypes.createStructField("age", DataTypes.IntegerType, true)); StructType structType = DataTypes.createStructType(structFiled); //转换为DataFrame DataFrame stuDF = sqlContext.createDataFrame(stuRow, structType); //注册临时表 stuDF.registerTempTable("students"); //执行sql语句 DataFrame tenagerDF = sqlContext.sql("select * from students where age<=18"); //FataFrame转换为RDD JavaRDD<Row> rowRDD = tenagerDF.javaRDD(); List<Row> row = rowRDD.collect(); 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 /** * @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("C://Users//Administrator//Desktop//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) } }