SparkSQL -DataFrame与RDD的互转

创建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) }    
}