Spark SQL与DataFrame原理解析与实现 spark研习第四季

二、 SparkSQL与DataFrame原理解析与实现

1. SparkSQL与DataFrame

SparkSQL之所以是除了SparkCore外最大的和最受关注的组件，原因是：

A）处理一切存储介质和各种格式的数据（同时可以方便地扩展SparkSQL的功能来支持更多类型的数据，例如Kudo，Kudo在存储和计算效率间取得了完美的平衡），包括实时数据处理。

2015年Hadoop出现了一种新的存储：Kudu

大数据9大神器之一：Kudo（大数据的奥秘电视节目中有讲解）

B）SparkSQL把数据仓库的计算能力推向了新的高度。不仅有无敌的计算速度（SparkSQL比Shark快了至少一个数量级，而Shark比Hive快了至少一个数量级。尤其是在Tungsten成熟以后会更加无可匹敌）。更为重要的是把数据仓库的计算复杂度推向了历史上全新的高度（SparkSQL后续推出的DataFrame可以让数据仓库直接使用机器学习图计算等复杂的算法库来对数据仓库进行深度数据价值挖掘）

Hive只是进行数据多维度查询。SparkSQL可以进行机器学习、图计算，所以是里程碑式的技术。

Shark为什么会被放弃？

=> Shark依赖的Hive不是官方原生版本，版本维护非常麻烦，Hive官方版本更新慢，所以决定中断Shark开发。

C）SparkSQ(DataFrame、DataSet）不仅是数据仓库的引擎，也是数据挖掘的引擎，更为重要的是SparkSQL是数据科学计算和分析引擎！！！

D） 后来的DataFrame让Spark（SQL）一举成为大数据计算引擎的技术实现霸主（尤其是在Tungsten的强力支持下）

传统数据库仅剩的应用场景：实时事务性分析。

数据库的死穴是扩展性差。数据库也不能分布式计算，

E）Hive+SparkSQL+DataFrame是目前至少在中国所有的大数据项目至少90%无法逃脱该技术组合，

- Hive负责廉价的数据仓库存储

- SparkSQL负责高速计算

- DataFrame负责复杂的数据挖掘

DataFrame是一个新的API

2、DataFrame与RDD

1） R和Python中都有DataFrame，Spark中的DataFrame从形式上看最大的不同点是其天生是分布式的。可以简单认为Spark中的DaraFrame是一个分布式的Table。形式如下所述：

Name    Age Tel

String  Int Long

String  Int Long

String  Int Long

...

String  Int Long

String  Int Long

String  Int Long

[/code]

而RDD是形如以下所述：

Person

Person

Person

...

Person

Person

Person

[/code]

DataFrame的每一列都有名称和类型。

2) RDD和DataFrame的根本差异：

A）RDD是以Record为单位的，Spark在优化时无法洞悉Record内部细节，所以也就无法进行更深度的优化，这极大限制了SparkSQL性能的提升！

B）DataFrame包含了每个Record的MetaData信息，即DataFrame的优化是基于列内部的优化，而不是像RDD一样只能基于行进行优化。

Hive包含元数据和数据本身，DataFrame也包含元数据和数据本身。

3) SparkSQL企业级最佳实践

第一阶段：最开始阶段，文件存储在FileSystem中，用c代码处理数据。

第二阶段：javaEE加数据库。

瓶颈：数据库不能分布式（扩展性），企业只能处理部分数据，数据需要经过过滤后再放进数据库中。

第三阶段：Hive。数据库局限太多导致转向hive，但hive计算能力有限，速度性能问题

第四阶段：hive 转向sparksql+Hive，计算能力是一个问题

第五阶段：hive+SparkSQL+dataFrame

第六阶段：hive+SparkSQL+dataFrame+dataSet

3. 使用Java和Scala两种语言实战DataFrame

创建DataFrame的时候，DataFrame可以来源于其他RDD，也可以来自于Hive表，或者其他数据来源。一般基于数据来源直接构造DataFrame。例如JSON文件，那么读取JSON文件的时候就会自动创建DataFrame。

SQLContext操作数据SQL的时候：有一个弊端就是，只支持SQL一种方言。

但是如果使用HiveContext的时候就可以支持不同种方言。

DataFrame的创建方式：

1. 可以来自己RDD

2. 来自Hive表

3. 其他数据来源（JSON等文件）

cat examples/src/main/resources/people.json

{"name":"Michael"}

{"name":"Andy", "age":30}

{"name":"Justin", "age":19}

//上传到HDFS中

hdfs dfs -put examples/src/main/resources/people.json /user/

[/code]

Java版本代码如下：

package com.dt.sparksql;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.sql.DataFrame;

import org.apache.spark.sql.SQLContext;

public class DataFramesql {

public static void main(String[] args){

//创建SparkConf用于读取系统配置信息并设置当前应用程序名称

SparkConf conf = new SparkConf().setAppName("DataFrameOps");

//创建JavaSparkContext对象实例作为整个Driver核心基石

JavaSparkContext sc = new JavaSparkContext(conf);

//创建SQLContext上下文对象用于SQL分析

SQLContext sqlContext = new  SQLContext(sc);

//创建DataFrame,可以简答的认为DataFrame是一张表

//DataFrame可以来源于多种格式，如Json，可以直接穿件DataFrame

//SQLContext只支持SQL一种方言，用HiveContext就可以支持多种方言（默认HQL，可以设置）

DataFrame df = sqlContext.read().json("HDFS://master:9000/user/people.json");

//select * from table;

df.show();

//describe table;

df.printSchema();

df.select("name").show();

//select name, age+1 from table;

df.select(df.col("name"), df.col("age").plus(10)).show();

//select * from table where age > 20;

df.filter(df.col("age").gt(20)).show();

//select count(1) from table group by age;

df.groupBy(df.col("age")).count().show();

}

}

[/code]

打包成.jar文件上传到集群中，写shell脚本：

cat sparkshell.sh

/opt/spark-1.4.0-bin-hadoop2.6/bin/spark-submit --class com.dt.sparksql.DataFramesql --master spark://master:7077 /opt/spark-1.4.0-bin-hadoop2.6/dataframesql.jar

[/code]

生产环境下，执行shell一般是这样的：

/opt/spark-1.4.0-bin-hadoop2.6/bin/spark-submit --class com.dt.sparksql.DataFramesql --files /opt/spark-1.4.0-bin-hadoop2.6/conf/hive-site.xml --driver-class-path /opt/hive-1.2.1/lib/mysql-connector-java-5.1.39-bin.jar  --master spark://master:7077 /opt/spark-1.4.0-bin-hadoop2.6/dataframesql.jar

[/code]

结果示例：

+----+-----+

| age|count|

+----+-----+

|null|    1|

|  19|    1|

|  30|    1|

+----+-----+

[/code]

Scala版本代码如下：

package com.dt.spark.sql

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

import org.apache.spark.sql.SQLContext

object DataFrameOps {

def main(args:Array[String]):Unit={

val conf = new SparkConf().setMaster("cluster").setAppName("DataFrameOps")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val df = sqlContext.read.json("HDFS://master:9000/user/people.json")

df.show()

df.printSchema()

df.select("name").show()

df.select(df.col("name"),df.col("age").plus(10)).show()

//  df.filter($"age" > 20).select($"name", $"age").show()

df.filter(df.col("age").gt(10)).show()

//df.groupBy("age").count().show()

df.groupBy(df.col("age")).count().show()

[/code]

4. RDD与DataFrame转换操作

1）RDD与DataFrame转换的重大意义

在Spark中RDD可以直接转换成DataFrame。SparkCore的核心是RDD，所有的调度都是基于RDD完成的，对RDD的操作都可以转换成基于DataFrame使用SparkSQL来操作。RDD可能接上数据库，接上NoSQL，其他文件系统等各种数据来源，然后将数据转换为DataFrame，

极大简化了大数据的开发，原来写Scala\Java，现在只需要写SparkSQL。

同时对DataFrame的操作又可以转换成RDD，基于DataFrame对数据进行SQL或机器学习等操作后又可以转换为RDD，这对于保存数据、格式化非常方便。

RDD变DataFrame有两种方式：

1.通过反射，推断RDD元素中的元数据。

RDD中的数据本身是没有元数据的，例如一个Person的信息里有id/name/age，RDD的Record不知道id/name/age这些信息，但如果变成DataFrame的话，DataFrame必须知道这些信息。如何在RDD和DataFrame转换时拥有这些元数据信息呢？最简单的就是通过反射。

在Scala中就是Case Class映射。写一个Case Class，描述RDD中不同列的元数据是什么。

在Java中就是通过JavaBean。

Scala：case class映射。

Java：Bean（但不能支持嵌套的JavaBean，也不能有List/Map等复杂的数据结构。只能用简单的数据类型：String/Int等。Scala就没有这些限制）

使用反射的前提：已经知道元数据信息了（静态的）。但有些场景下只有在运行时才能知道元数据信息（动态的）

2、创建DataFrame时事先不知道元数据信息，只能在运行时动态构建元数据。然后再把这些元数据信息应用于RDD上。这种情况是比较常见的情况，即动态获取Schema。

1） 通过反射的方式来推断RDD元素中的元数据。因为RDD本身一条数据本身是没有元数据的，例如Person，而Person有name,id等，而record是不知道这些的，但是变成DataFrame背后一定知道，通过反射的方式就可以了解到背后这些元数据，进而转换成DataFrame。

如何反射？

Scala: 通过case class映射，在case class里面说我们这个RDD里面每个record的不同列的元数据是什么。

Java: 如何描述数据的元数据？构建Java Bean，使用Java Bean构建元数据信息，然后变换成DataFrame，但是此种方法不可以构建DataFrame嵌套类型。

2）动态获取Schema,我们并不知道RDD的元数据信息，所以只能根据曾经运行时动态构建一份具体的元数据。然后将具体的元数据运行在存在的RDD上。而且这种情况比较常见。

3）代码实战

package com.dt.spark.SparkApps.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

Person [id=1, name=Spark, age=7]

Person [id=2, name=Hadoop, age=10]

*/

public class RDDToDataFrameByReflection {

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local").setAppName("RDDToDataFrameByReflection");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

//读取数据

JavaRDD<String> lines = sc.textFile("G://datarguru spark//tool//persons.txt");

JavaRDD<Person> persons = lines.map(new Function<String,Person>(){

private static final long serialVersionUID = 1L;

@Override

public Person call(String line) throws Exception {

String[] splited = line.split(",");

Person p = new Person();

p.setId(Integer.valueOf(splited[0].trim()));

p.setName(splited[1].trim());

p.setAge(Integer.valueOf(splited[2].trim()));

return p;

}

});

//第一个参数：RDD，第二个参数是JavaBean，Person类

//第二参数就是封装的JavaBean，JavaBean中封装了Person的元数据信息，

//通过第二个参数DataFrame也就获得了元数据信息。

//在底层通过反射的方式获得Person的所有Fields，结合RDD本身，就生成了DataFrame

DataFrame df = sqlContext.createDataFrame(persons, Person.class);

df.registerTempTable("persons");

DataFrame bigDatas = sqlContext.sql("select * from persons where age >= 6");

//DataFrame => RDD

JavaRDD<Row> bigDataRDD = bigDatas.javaRDD();

JavaRDD<Person> result = bigDataRDD.map(new Function<Row,Person>(){

private static final long serialVersionUID = 1L;

@Override

public Person call(Row row) throws Exception {

//返回具体每条记录

Person p = new Person();

/**

* 由于数据在DataFrame会进行优化，里面会对元数据进行排序

* 顺序可能就不是id name age的顺序了。

      */

p.setId(row.getInt(1));

p.setName(row.getString(2));

p.setAge(row.getInt(0));

return p;

}

});

List<Person> personList = result.collect();

for(Person p : personList){

System.out.println(p);

}

}

}

[/code]

People.java源码如下：

package com.dt.spark.SparkApps.sql;

import java.io.Serializable;

//因为底层是反射，要求JavaBean是public

//此时需要序列化，因为是分布式方式。

public class Person implements Serializable{

private static final long serialVersionUID = 1L;

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 "Person [id=" + id + ", name=" + name + ", age=" + age + "]";

}

}

[/code]

如果出现错误：

ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)

java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String

at org.apache.spark.sql.Row$class.getString(Row.scala:250)

at org.apache.spark.sql.catalyst.expressions.GenericRow.getString(rows.scala:192)

at com.dt.sparksql.RDDToDataFrameByReflection$2.call(RDDToDataFrameByReflection.java:56)

at com.dt.sparksql.RDDToDataFrameByReflection$2.call(RDDToDataFrameByReflection.java:1)

at org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1.apply(JavaPairRDD.scala:1015)

//原因是隐式转换的问题，因为DataFrame有自己的优化方式，转换之后并一定是不读入时候的顺序

Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String

at org.apache.spark.sql.Row$class.getString(Row.scala:250)

[/code]

5. RDD和DataFrame动态转换操作

1）非动态转换和动态转换

什么是非动态转换？

=> 提前已经知道了RDD具体数据的元数据信息，可以通过JavaBean或Case Class的方式提前创建DataFrame时，通过反射的方式获得元数据信息。

什么是动态转换？

=> 无法提前知道具体的RDD每个Record的列的个数及每列的类型只有在运行时才能知道。

这种情况在生产环境下更常见。因为在生产环境下提前知道数据的元数据信息的可能性不大。另外，生产环境下业务会变化，业务变化时列就会变化，或列的数据类型变化 ，如果采用静态的方式的话，每次业务变化时代码需要大幅度的修改。如果采用动态的方式，元数据来自于数据库或文件或其他方式，业务变化时代码的业务逻辑不会发生过多变化。

Scala实现的两种方式：

1 在main方法中

2 继承app去写代码。

2）Java实现RDD与DataFrame的动态转换

package com.dt.spark.SparkApps.sql;

/**

* 打印结果：

[1,Spark,7]

[2,Hadoop,10]

[3,Flink,5]

*/

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;

public class RDDToDataFrameByProgrammatically {

public static void main(String[] args) {

SparkConf conf = new SparkConf().setMaster("local").

setAppName("RDDToDataFrameByReflection");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

//读取数据

JavaRDD<String> lines = sc.textFile("E://persons.txt");

/**

* 第一步：在RDD的基础上创建类型为Row的RDD

       */

//首先，必须将RDD变成以Row为类型的RDD。Row可以简单理解为Table的一行数据

JavaRDD<Row> personsRDD = lines.map(new Function<String,Row>(){

@Override

public Row call(String line) throws Exception {

String[] splited = line.split(",");

return RowFactory.create(Integer.valueOf(splited[0]),splited[1],Integer.valueOf(splited[2]));

}

});

/**

* 第二步：动态构造DataFrame的元数据，一般而言，有多少列以及每列的具体类型可能来自于

* JSON文件，也可能来自于DB

       */

//对Row具体指定元数据信息。

List<StructField> structFields = new ArrayList<StructField>();

//列名称  列的具体类型（Integer Or String） 是否为空一般为true，实际在开发环境是通过for循环，而不是手动添加

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,用于最后DataFrame元数据的描述

StructType structType = DataTypes.createStructType(structFields);

/**

* 第三步：基于已有的MetaData以及RDD<Row>来构造DataFrame

       */

DataFrame personsDF = sqlContext.createDataFrame(personsRDD, structType);

/**

* 第四步：注册成临时表以供后续的SQL查询操作

       */

personsDF.registerTempTable("persons");

/**

* 第五步：进行数据的多维度分析

       */

DataFrame result = sqlContext.sql("select * from persons");

/**

* 第六步：对结果进行处理，包括由DataFrame转换成为RDD<Row>,以及结果的持久化

       */

List<Row> listRow = result.javaRDD().collect();

for(Row row : listRow){

System.out.println(row);

}

}

}

[/code]