一、前述

      1、SparkSQL介绍

Hive是Shark的前身,Shark是SparkSQL的前身,SparkSQL产生的根本原因是其完全脱离了Hive的限制。

  • SparkSQL支持查询原生的RDD。 RDD是Spark平台的核心概念,是Spark能够高效的处理大数据的各种场景的基础。
  • 能够在Scala中写SQL语句。支持简单的SQL语法检查,能够在Scala中写Hive语句访问Hive数据,并将结果取回作为RDD使用。

    2、Spark on Hive和Hive on Spark

  • Spark on Hive: Hive只作为储存角色,Spark负责sql解析优化,执行。

    Hive on Spark:Hive即作为存储又负责sql的解析优化,Spark负责执行。

二、基础概念

         1、DataFrame

DataFrame也是一个分布式数据容器。与RDD类似,然而DataFrame更像传统数据库的二维表格,除了数据以外,还掌握数据的结构信息,即schema同时,与Hive类似,DataFrame也支持嵌套数据类型(struct、array和map)。从API易用性的角度上 看, DataFrame API提供的是一套高层的关系操作,比函数式的RDD API要更加友好,门槛更低。

DataFrame的底层封装的是RDD,只不过RDD的泛型是Row类型。

     2、SparkSQL的数据源

SparkSQL的数据源可以是JSON类型的字符串,JDBC,Parquent,Hive,HDFS等。

    3、SparkSQL底层架构

首先拿到sql后解析一批未被解决的逻辑计划,再经过分析得到分析后的逻辑计划,再经过一批优化规则转换成一批最佳优化的逻辑计划,再经过SparkPlanner的策略转化成一批物理计划,随后经过消费模型转换成一个个的Spark任务执行。

    4、谓词下推(predicate Pushdown)

 

三。创建DataFrame的几种方式

  1、读取json格式的文件创建DataFrame

  • json文件中的json数据不能嵌套json格式数据。
  • DataFrame是一个一个Row类型的RDD,df.rdd()/df.javaRdd()。
  • 可以两种方式读取json格式的文件。
  • df.show()默认显示前20行数据。
  • DataFrame原生API可以操作DataFrame(不方便)。
  • 注册成临时表时,表中的列默认按ascii顺序显示列。

java代码:

SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("jsonfile");
SparkContext sc = new SparkContext(conf); //创建sqlContext
SQLContext sqlContext = new SQLContext(sc);//SprakSQL中是SQLContext对象 /**
* DataFrame的底层是一个一个的RDD RDD的泛型是Row类型。
* 以下两种方式都可以读取json格式的文件
*/
DataFrame df = sqlContext.read().format("json").load("sparksql/json");
// DataFrame df2 = sqlContext.read().json("sparksql/json.txt");
// df2.show();
/**
* DataFrame转换成RDD
*/
RDD<Row> rdd = df.rdd();
/**
* 显示 DataFrame中的内容,默认显示前20行。如果现实多行要指定多少行show(行数)
* 注意:当有多个列时,显示的列先后顺序是按列的ascii码先后显示。
*/
// df.show();
/**
* 树形的形式显示schema信息
*/
df.printSchema(); /**
* dataFram自带的API 操作DataFrame(
很麻烦)
*/
//select name from table
// df.select("name").show();
//select name age+10 as addage from table
df.select(df.col("name"),df.col("age").plus(10).alias("addage")).show();
//select name ,age from table where age>19
df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show();
//select count(*) from table group by age
df.groupBy(df.col("age")).count().show(); /**
* 将DataFrame注册成临时的一张表,这张表临时注册到内存中,是逻辑上的表,不会雾化到磁盘
*/
df.registerTempTable("jtable"); DataFrame sql = sqlContext.sql("select age,count(1) from jtable group by age");
DataFrame sql2 = sqlContext.sql("select * from jtable"); sc.stop();

scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("jsonfile") val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.json("sparksql/json")
//val df1 = sqlContext.read.format("json").load("sparksql/json") df.show()
df.printSchema()
//select * from table
df.select(df.col("name")).show()
//select name from table where age>19
df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show()
//select count(*) from table group by age
df.groupBy(df.col("age")).count().show(); /**
* 注册临时表
*/
df.registerTempTable("jtable")
val result = sqlContext.sql("select * from jtable")
result.show()
sc.stop()

2、通过json格式的RDD创建DataFrame

 java代码:

SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("jsonRDD");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList(
"{\"name\":\"zhangsan\",\"age\":\"18\"}",
"{\"name\":\"lisi\",\"age\":\"19\"}",
"{\"name\":\"wangwu\",\"age\":\"20\"}"
));
JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList(
"{\"name\":\"zhangsan\",\"score\":\"100\"}",
"{\"name\":\"lisi\",\"score\":\"200\"}",
"{\"name\":\"wangwu\",\"score\":\"300\"}"
)); DataFrame namedf = sqlContext.read().json(nameRDD);
DataFrame scoredf = sqlContext.read().json(scoreRDD);
namedf.registerTempTable("name");
scoredf.registerTempTable("score"); DataFrame result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name");
result.show(); sc.stop();

scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("jsonrdd")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc) val nameRDD = sc.makeRDD(Array(
"{\"name\":\"zhangsan\",\"age\":18}",
"{\"name\":\"lisi\",\"age\":19}",
"{\"name\":\"wangwu\",\"age\":20}"
))
val scoreRDD = sc.makeRDD(Array(
"{\"name\":\"zhangsan\",\"score\":100}",
"{\"name\":\"lisi\",\"score\":200}",
"{\"name\":\"wangwu\",\"score\":300}"
))
val nameDF = sqlContext.read.json(nameRDD)
val scoreDF = sqlContext.read.json(scoreRDD)
nameDF.registerTempTable("name")
scoreDF.registerTempTable("score")
val result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name")
result.show()
sc.stop()

3、非json格式的RDD创建DataFrame(重要)

1) 通过反射的方式将非json格式的RDD转换成DataFrame(不建议使用)

  • 自定义类要可序列化
  • 自定义类的访问级别是Public
  • RDD转成DataFrame后会根据映射将字段按Assci码排序
  • 将DataFrame转换成RDD时获取字段两种方式,一种是df.getInt(0)下标获取(不推荐使用),另一种是df.getAs(“列名”)获取(推荐使用)
  • 关于序列化问题:

              1.反序列化时serializable 版本号不一致时会导致不能反序列化。
              2.子类中实现了serializable接口,父类中没有实现,父类中的变量不能被序列化,序列化后父类中的变量会得到null。
              注意:父类实现serializable接口,子类没有实现serializable接口时,子类可以正常序列化
              3.被关键字transient修饰的变量不能被序列化。
              4.静态变量不能被序列化,属于类,不属于方法和对象,所以不能被序列化。
             另外:一个文件多次writeObject时,如果有相同的对象已经写入文件,那么下次再写入时,只保存第二次写入的引用,读取时,都是第一次保存的对象。

java代码:

/**
* 注意:
* 1.自定义类必须是可序列化的
* 2.自定义类访问级别必须是Public
* 3.RDD转成DataFrame会把自定义类中字段的名称按assci码排序
*/
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("RDD");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile("sparksql/person.txt");
JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public Person call(String s) throws Exception {
Person p = new Person();
p.setId(s.split(",")[0]);
p.setName(s.split(",")[1]);
p.setAge(Integer.valueOf(s.split(",")[2]));
return p;
}
});
/**
* 传入进去Person.class的时候,sqlContext是通过反射的方式创建DataFrame
* 在底层通过反射的方式获得Person的所有field,结合RDD本身,就生成了DataFrame
*/
DataFrame df = sqlContext.createDataFrame(personRDD, Person.class);
df.show();
df.registerTempTable("person");
sqlContext.sql("select name from person where id = 2").show(); /**
* 将DataFrame转成JavaRDD
* 注意:
* 1.可以使用row.getInt(0),row.getString(1)...通过下标获取返回Row类型的数据,但是要注意列顺序问题---不常用
* 2.可以使用row.getAs("列名")来获取对应的列值。
*
*/
JavaRDD<Row> javaRDD = df.javaRDD();
JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public Person call(Row row) throws Exception {
Person p = new Person();
//p.setId(row.getString(1));
//p.setName(row.getString(2));
//p.setAge(row.getInt(0)); p.setId((String)row.getAs("id"));
p.setName((String)row.getAs("name"));
p.setAge((Integer)row.getAs("age"));
return p;
}
});
map.foreach(new VoidFunction<Person>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public void call(Person t) throws Exception {
System.out.println(t);
}
}); sc.stop();

scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("rddreflect")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile("./sparksql/person.txt")
/**
* 将RDD隐式转换成DataFrame
*/
import sqlContext.implicits._ val personRDD = lineRDD.map { x => {
val person = Person(x.split(",")(0),x.split(",")(1),Integer.valueOf(x.split(",")(2)))
person
} }
val df = personRDD.toDF();
df.show() /**
* 将DataFrame转换成PersonRDD
*/
val rdd = df.rdd
val result = rdd.map { x => {
Person(x.getAs("id"),x.getAs("name"),x.getAs("age"))
} }
result.foreach { println}
sc.stop()

结果:

1) 动态创建Schema将非json格式的RDD转换成DataFrame(建议使用)

java:

SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("rddStruct");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt");
/**
* 转换成Row类型的RDD
*/
JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public Row call(String s) throws Exception {
return RowFactory.create(//这里字段顺序一定要和下边 StructField对应起来
String.valueOf(s.split(",")[0]),
String.valueOf(s.split(",")[1]),
Integer.valueOf(s.split(",")[2])
);
}
});
/**
* 动态构建DataFrame中的元数据,一般来说这里的字段可以来源自字符串,也可以来源于外部数据库
*/
List<StructField> asList =Arrays.asList(//这里字段顺序一定要和上边对应起来
DataTypes.createStructField("id", DataTypes.StringType, true),
DataTypes.createStructField("name", DataTypes.StringType, true),
DataTypes.createStructField("age", DataTypes.IntegerType, true)
); StructType schema = DataTypes.createStructType(asList);
DataFrame df = sqlContext.createDataFrame(rowRDD, schema); df.show();
    JavaRDD<Row> javaRDD = df.javaRDD();
javaRDD.foreach(new VoidFunction<Row>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public void call(Row row) throws Exception {//Row类型的RDD
System.out.println(row.getString(0));
}
})
sc.stop();

scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("rddStruct")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile("./sparksql/person.txt")
val rowRDD = lineRDD.map { x => {
val split = x.split(",")
RowFactory.create(split(0),split(1),Integer.valueOf(split(2)))
} } val schema = StructType(List(
StructField("id",StringType,true),
StructField("name",StringType,true),
StructField("age",IntegerType,true)
)) val df = sqlContext.createDataFrame(rowRDD, schema)
df.show()
df.printSchema()
sc.stop()

 

4、读取parquet文件创建DataFrame

注意:

  • 可以将DataFrame存储成parquet文件。保存成parquet文件的方式有两种
  • df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");
    df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");
  • SaveMode指定文件保存时的模式。

Overwrite:覆盖

Append:追加

ErrorIfExists:如果存在就报错

Ignore:如果存在就忽略

java代码:

SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("parquet");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> jsonRDD = sc.textFile("sparksql/json");
DataFrame df = sqlContext.read().json(jsonRDD);
/**
* 将DataFrame保存成parquet文件,SaveMode指定存储文件时的保存模式
* 保存成parquet文件有以下两种方式:
*/
df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");
df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");
df.show();
/**
* 加载parquet文件成DataFrame
* 加载parquet文件有以下两种方式:
*/ DataFrame load = sqlContext.read().format("parquet").load("./sparksql/parquet");
load = sqlContext.read().parquet("./sparksql/parquet");
load.show(); sc.stop()

scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("parquet")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val jsonRDD = sc.textFile("sparksql/json")
val df = sqlContext.read.json(jsonRDD)
df.show()
/**
* 将DF保存为parquet文件
*/
df.write.mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet")
df.write.mode(SaveMode.Overwrite).parquet("./sparksql/parquet")
/**
* 读取parquet文件
*/
var result = sqlContext.read.parquet("./sparksql/parquet")
result = sqlContext.read.format("parquet").load("./sparksql/parquet")
result.show()
sc.stop()

5、读取JDBC中的数据创建DataFrame(MySql为例)

两种方式创建DataFrame

java代码:

SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("mysql");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
/**
* 第一种方式读取MySql数据库表,加载为DataFrame
*/
Map<String, String> options = new HashMap<String,String>();
options.put("url", "jdbc:mysql://192.168.179.4:3306/spark");
options.put("driver", "com.mysql.jdbc.Driver");
options.put("user", "root");
options.put("password", "123456");
options.put("dbtable", "person"
);
DataFrame person = sqlContext.read().format("jdbc").options(options).load();
person.show();
person.registerTempTable("person");
/**
* 第二种方式读取MySql数据表加载为DataFrame
*/
DataFrameReader reader = sqlContext.read().format("jdbc");
reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark");
reader.option("driver", "com.mysql.jdbc.Driver");
reader.option("user", "root");
reader.option("password", "123456");
reader.option("dbtable", "score"
);
DataFrame score = reader.load();
score.show();
score.registerTempTable("score"); DataFrame result =
sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name");
result.show();
/**
* 将DataFrame结果保存到Mysql中
*/
Properties properties = new Properties();
properties.setProperty("user", "root");
properties.setProperty("password", "123456");
result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties); sc.stop();

 scala代码:

val conf = new SparkConf()
conf.setMaster("local").setAppName("mysql")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
/**
* 第一种方式读取Mysql数据库表创建DF
*/
val options = new HashMap[String,String]();
options.put("url", "jdbc:mysql://192.168.179.4:3306/spark")
options.put("driver","com.mysql.jdbc.Driver")
options.put("user","root")
options.put("password", "123456")
options.put("dbtable","person")

val person = sqlContext.read.format("jdbc").options(options).load()
person.show()
person.registerTempTable("person")
/**
* 第二种方式读取Mysql数据库表创建DF
*/
val reader = sqlContext.read.format("jdbc")
reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark")
reader.option("driver","com.mysql.jdbc.Driver")
reader.option("user","root")
reader.option("password","123456")
reader.option("dbtable", "score")

val score = reader.load()
score.show()
score.registerTempTable("score")
val result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name")
result.show()
/**
* 将数据写入到Mysql表中
*/
val properties = new Properties()
properties.setProperty("user", "root")
properties.setProperty("password", "123456")
result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties) sc.stop()

【Spark篇】---SparkSQL初始和创建DataFrame的几种方式的更多相关文章

  1. JAVA SparkSQL初始和创建DataFrame的几种方式

    建议参考SparkSQL官方文档:http://spark.apache.org/docs/latest/sql-programming-guide.html 一.前述       1.SparkSQ ...

  2. Spark SQL初始化和创建DataFrame的几种方式

    一.前述       1.SparkSQL介绍 Hive是Shark的前身,Shark是SparkSQL的前身,SparkSQL产生的根本原因是其完全脱离了Hive的限制. SparkSQL支持查询原 ...

  3. Spark 基础 —— 创建 DataFrame 的三种方式

    1.自定义 schema(Rdd[Row] => DataSet[Row]) import org.apache.spark.sql.types._ val peopleRDD = spark. ...

  4. sparkSQL获取DataFrame的几种方式

    sparkSQL获取DataFrame的几种方式 1. on a specific DataFrame. import org.apache.spark.sql.Column df("col ...

  5. Pandas 基础(3) - 生成 Dataframe 的几种方式

    这一节想总结一下 生成 Dataframe 的几种方式: CSV Excel python dictionary List of tuples List of dictionary 下面分别一一介绍具 ...

  6. 7 -- Spring的基本用法 -- 7... 创建Bean的3种方式

    7.7 创建Bean的3种方式 ① 调用构造器创建Bean. ② 调用静态工厂方法创建Bean. ③ 调用实例工厂方法创建Bean. 7.7.1 使用构造器创建Bean实例. 使用构造器来创建Bean ...

  7. 当阿里面试官问我:Java创建线程有几种方式?我就知道问题没那么简单

    这是最新的大厂面试系列,还原真实场景,提炼出知识点分享给大家. 点赞再看,养成习惯~ 微信搜索[武哥聊编程],关注这个 Java 菜鸟. 昨天有个小伙伴去阿里面试实习生岗位,面试官问他了一个老生常谈的 ...

  8. 0036 Java学习笔记-多线程-创建线程的三种方式

    创建线程 创建线程的三种方式: 继承java.lang.Thread 实现java.lang.Runnable接口 实现java.util.concurrent.Callable接口 所有的线程对象都 ...

  9. 【java并发】传统线程技术中创建线程的两种方式

    传统的线程技术中有两种创建线程的方式:一是继承Thread类,并重写run()方法:二是实现Runnable接口,覆盖接口中的run()方法,并把Runnable接口的实现扔给Thread.这两种方式 ...

随机推荐

  1. Python—变量

    1.在Python中,变量名类似__xxx__的,也就是以双下划线开头,并且以双下划线结尾的,是特殊变量,特殊变量是可以直接访问的,不是private变量 2.访问限制: class内部属性可以被外部 ...

  2. hibernate(九)多对多关联

    原文链接:http://www.orlion.ml/29/ 一.多对多单向关联 假设一个老师教多个学生,一个学生被多个老师教,这就是典型的多对多关系 配置方式是在Teacher类的getStudent ...

  3. Xcode中iPhone iPad模拟器调整大小的方法

    Xcode中调试iPad程序默认的iPad模拟器非常小,如何方法iPad模拟器的显示尺寸呢? 选中iOS模拟器,在“Window -> 缩放比例”中就可以调整了. 快捷键: Command + ...

  4. sonar runner 2.4

    https://www.versioneye.com/java/org.codehaus.sonar.runner:sonar-runner-dist/2.4

  5. 遇到java.lang.OutOfMemoryError: Java heap space问题【持续跟踪中...】

    昨天下午app服务端无缘无故又502错了,我打开tomcat 日志: INFO: org.tuckey.web.filters.urlrewrite.UrlRewriteFilter DEBUG: i ...

  6. 一篇%3CDIV%20style%3D%22FONT-SIZE%

    %3CDIV%20style%3D%22FONT-SIZE%3A%2016px%22%3E1%EF%BC%8C%E6%88%91%E4%BB%A5%E4%B8%BA%E7%BB%88%E6%9C%89 ...

  7. @ResponseBody注解

    作用 @ResponseBody注解表示该方法的返回结果直接写入HTTP response body中 原理 在使用此注解之后跳过视图处理器,将返回的对象通过适当的转换器转换为指定的格式之后,直接将数 ...

  8. 使用Jenkins+gitlab自动化构建时排除分支

    我们的目的是gitlab上的代码有变动时会自动向Jenkins发送web钩子请求,触发指定的动作: 但默认情况下,所有分支(如测试环境和预生产)的代码有变动时都会触发,此时可以在Jenkins的项目设 ...

  9. Luogu P3919【模板】可持久化数组(可持久化线段树/平衡树)

    题面:[模板]可持久化数组(可持久化线段树/平衡树) 不知道说啥,总之我挺喜欢自己打的板子的! #include<cstdio> #include<cstring> #incl ...

  10. HPU第三次积分赛-D:Longest Increasing Subsequence(DP)

    Longest Increasing Subsequence 描述 给出一组长度为n的序列,a1​,a2​,a3​,a4​...an​, 求出这个序列长度为k的严格递增子序列的个数 输入 第一行输入T ...