spark sql

code/spark/spark-core/pom.xml

@@ -10,6 +10,7 @@
 
     <properties>
         <scala.version>2.12</scala.version>
+        <spark.version>2.4.0</spark.version>
     </properties>
 
     <build>
@@ -31,8 +32,14 @@
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_${scala.version}</artifactId>
-            <version>2.4.0</version>
+            <version>${spark.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.version}</artifactId>
+            <version>${spark.version}</version>
+        </dependency>
+
         <!--单元测试依赖包-->
         <dependency>
             <groupId>junit</groupId>

code/spark/spark-core/src/main/java/rdd/scala/SparkSqlTest.scala

@@ -0,0 +1,57 @@
+package rdd.scala
+
+import org.apache.spark.sql.{Dataset, SparkSession}
+
+
+object SparkSqlTest extends App {
+
+
+
+  val spark = SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()
+
+  val dataFrames = spark.read.json("/usr/file/people.json")
+
+  df.select("name").show()
+
+  df.printSchema()
+
+
+
+
+  import spark.implicits._
+
+  val primitiveDS = Seq(1, 2, 3).toDS()
+  primitiveDS.printSchema()
+  primitiveDS.map(_ + 1).collect()
+
+  peopleDS.select("name").show()  //失败
+  peopleDS.dtypes
+  peopleDS.printSchema()
+  peopleDS.toDF()
+  // Encoders are created for case classes
+
+  /* 1.此时把selected写成为selected ,编译器没有任何提示 */
+  spark.sql("selected name from emp")
+
+  /* 2.此时把selected写成为selected ,编译器有提示; 但是把字段名称name写成了nameEd ,编译器没有任何提示*/
+  val dataFrames = spark.read.json("people.json")
+  dataFrames.selected("nameEd").show()
+  dataFrames.map(line=>line.name)
+
+  case class Person(name: String, age: Long)
+
+  /* 3.此时最为严格,语法和字段名称错误都被检测出来*/
+  val dataSet: Dataset[Person] = spark.read.json("people.json").as[Person]
+  dataSet.selected("name")
+  dataSet.map(line=>line.name)
+  dataSet.map(line=>line.nameEd)
+
+  /* 4.即使在由RDD转换为dataFrame时候指定了类型Person,依然无法提示字段名称*/
+  val peopleDF = spark.sparkContext
+    .textFile("people.json")
+    .map(_.split(","))
+    .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
+    .toDF()
+  peopleDF.map(line=>line.name)
+
+}

notes/SparkSQL-Datasets&DataFrames.md

@@ -0,0 +1,158 @@
+# Spark SQL DataFrames
+
+<nav>
+<a href="#一Spark-SQL简介">一、Spark SQL简介</a><br/>
+<a href="#二DataFrame--DataSet">二、DataFrame & DataSet </a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#21-DataFrame">2.1 DataFrame  </a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#22-DataFrame-对比-RDDs">2.2 DataFrame 对比 RDDs</a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#23-DataSet">2.3 DataSet</a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#24-静态类型与运行时类型安全">2.4 静态类型与运行时类型安全</a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#25-Untyped--Typed">2.5 Untyped & Typed </a><br/>
+<a href="#三DataFrame--DataSet---RDDs-总结">三、DataFrame & DataSet  & RDDs 总结</a><br/>
+<a href="#四Spark-SQL的运行原理">四、Spark SQL的运行原理</a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#41-逻辑计划Logical-Plan">4.1 逻辑计划(Logical Plan)</a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#42-物理计划Physical-Plan">4.2 物理计划(Physical Plan) </a><br/>
+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#43-执行">4.3 执行</a><br/>
+</nav>
+
+## 一、Spark SQL简介
+
+Spark SQL是Spark中的一个子模块，主要用于操作结构化数据。其具有以下特点：
+
++  能够将SQL查询与Spark程序无缝混合。允许您使用SQL或DataFrame API对结构化数据进行查询，支持Java，Scala，Python和R语言；
++ Spark SQL支持多种数据源，包括Hive，Avro，Parquet，ORC，JSON和JDBC；
++ Spark SQL支持HiveQL语法以及Hive SerDes和UDF，允许你访问现有的Hive仓库；
++ 支持标准的JDBC和ODBC连接；
++ Spark SQL支持优化器，列式存储和代码生成等特性，以提高查询效率；
++ 支持扩展并能保证容错。
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/sql-hive-arch.png"/> </div>
+
+## 二、DataFrame & DataSet 
+
+### 2.1 DataFrame  
+
+为了支持结构化数据的处理，Spark SQL提供了新的数据结构DataFrame。
+
+DataFrame是一个由具名列组成的数据集。它在概念上等同于关系数据库中的表或R/Python语言中的data frame，同时DataFrame还在底层做了更多的优化，以保证查询效率。 DataFrame具备以下特点：
+
++ 支持多种数据源，如结构化数据文件，Hive表，外部数据库或现有RDD；
++ 支持Scala，Java，Python和R语言，在Scala和Java中，DataFrame由行数据集(Rows)表示：
+  + 在Scala API中，DataFrame等价于Dataset [Row]；
+  + 在Java API中，DataFrame等价于Dataset\<Row>。
+
+| 语言   | 主要抽象                                     |
+| ------ | -------------------------------------------- |
+| Scala  | Dataset[T] & DataFrame (Dataset[Row] 的别名) |
+| Java   | Dataset[T]                                   |
+| Python | DataFrame                                    |
+| R      | DataFrame                                    |
+
+### 2.2 DataFrame 对比 RDDs
+
+DataFrame和RDDs最主要的区别在于一个面向的是结构化数据，一个面向的是非结构化数据，它们内部的数据结构如下：
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-dataFrame+RDDs.png"/> </div>
+
+DataFrame内部的有明确Scheme结构，即列名、列字段类型都是已知的，这带来的好处是可以减少数据读取以及更好地优化执行计划，从而保证查询效率。
+
+**DataFrame和RDDs应该如何选择？**
+
++ 如果你想使用函数式编程而不是DataFrame API，则使用RDDs；
++ 如果你的数据是非结构化的(比如流媒体或者字符流)，则使用RDDs，
++ 如果你的数据是结构化的(如RDBMS中的数据)或者半结构化的(如日志)，出于性能和优化上的考虑，建议使用DataFrame。
+
+### 2.3 DataSet
+
+在上一小节中，我们提到了Dataset这个概念，这里做一下解释：
+
+Dataset是分布式的数据集合，在Spark 1.6版本被引入。它集成了RDD和Spark SQL的优点，具备强类型，完善的lambda函数和执行引擎优化等特点，支持Scala和Java语言。在Spark 2.0后，为了方便开发者，Spark将DataFrame和Dataset的API融合到一起，提供一致性API，即用户可以通过一套标准API完成对两者的操作。
+
+> 这里注意一下：DataFrame被标记为Untyped API，而DataSet被标记为Typed API，后文会对两者做出解释。
+
+
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-unifed.png"/> </div>
+
+### 2.4 静态类型与运行时类型安全
+
+静态类型(Static-typing)与运行时类型安全(runtime type-safety) 主要表现如下:
+
+在实际使用中，如果你用的是Spark SQL的查询语句，则直到运行时你才会发现有语法错误，而如果你用的是DataFrame和 Dataset，则在编译时就可以发现错误(这节省了开发时间和整体代价)。DataFrame和Dataset主要区别在于：
+
+在DataFrame中，当你调用了API之外的函数，编译器就会报错，但如果你使用了一个不存在的字段名字，编译器依然无法发现。而在Dataset中，因为其API都是用lambda函数和JVM类型对象表示的，所有不匹配的类型参数在编译时就会被发现。
+
+以上这些最终都被解释成关于类型安全图谱，对应开发中的语法和分析错误。在图谱中，Dataset最严格，但对于开发者来说效率最高。
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-运行安全.png"/> </div>
+
+上面的描述可能并没有那么直观，下面的给出一个IDEA中代码编译的示例：
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-运行时类型安全.png"/> </div>
+
+这里一个可能的疑惑是DataFrame明明是有确定的Scheme结构(即列名、列字段类型都是已知的)，但是为什么还是无法对列名进行推断和错误判断，这是因为DataFrame是Untyped的。
+
+### 2.5 Untyped & Typed 
+
+在2.3小节，我们介绍过DataFrame API被标记为Untyped API，而DataSet API被标记为Typed API。
+
+DataFrame的Untyped是相对于语言(或API)层面而言，它确实有明确的Scheme结构，即列名，列类型都是确定的，但这些信息完全由Spark来维护，Spark只会在运行时检查这些类型和指定类型是否一致。这也就是为什么在Spark 2.0之后，官方推荐把DataFrame看做是`DatSet[Row]`，Row是spark中定义的一个`trait`，其子类中封装了列字段的信息。
+
+相对而言，DataSet是Typed的，即强类型。如下面代码，dataSet的类型由case class(scala)或者Java bean(Java)来明确指定的，在这里即每一行数据代表一个`Person`，这些信息由JVM来保证正确性，所以字段名错误和类型错误在编译的时候就会被IDE发现。
+
+```scala
+case class Person(name: String, age: Long)
+val dataSet: Dataset[Person] = spark.read.json("people.json").as[Person]
+```
+
+
+
+## 三、DataFrame & DataSet  & RDDs 总结
+
+这里对三者做一下简单的总结：
+
++ RDDs适合非结构化数据的处理，而DataFrame & DataSet更适合结构化数据和半结构化的处理；
++ DataFrame & DataSet可以通过统一的Dataset API进行访问，而RDDs则更适合函数式编程的场景；
++ 相比于DataFrame而言，DataSet是强类型的(Typed)，有着更为严格的静态类型检查；
++ DataSets、DataFrames、SQL的底层都依赖了RDDs API，并对外提供结构化的访问接口。
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-structure-api.png"/> </div>
+
+
+
+## 四、Spark SQL的运行原理
+
+对于结构化APIs(DataFrames & DataSets  & SQL)，其实际执行流程都是相同的：
+
+1.  进行DataFrame/Dataset/SQL编程；
+
+2. 如果是有效的代码，即代码没有编译错误，Spark 将其转换为一个逻辑计划；
+3. Spark 将此逻辑计划转换为物理计划，同时进行代码优化；
+4. Spark 然后在集群上执行这个物理计划(基于 RDD 操作) 。
+
+### 4.1 逻辑计划(Logical Plan)
+
+执行的第一个阶段是将用户代码转换成一个逻辑计划。它首先将用户代码转换成unresolved logical plan(未解决的逻辑计划)，之所以这个计划是未解决的，是因为尽管您的代码在语法上是正确的，但是它引用的表或列可能不存在。 Spark使用analyzer(分析器)基于catalog(存储的所有表和DataFrame信息)进行解析。解析失败则拒绝执行，解析成功则将结果传给Catalyst优化器(Catalyst Optimizer)，优化器是一组规则的集合，用于优化逻辑计划，通过谓词下推等方式进行优化，最终输出优化后的逻辑执行计划。
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-Logical-Planning.png"/> </div>
+
+
+
+### 4.2 物理计划(Physical Plan) 
+
+在得到优化后的逻辑计划后，Spark就开始了物理计划过程。 它如何通过生成不同的物理执行策略，并通过成本模型来比较它们，从而选择一个最优的物理计划在集群上面执行的。物理规划的输出结果是一系列的RDDs和转换关系(transformations)。
+
+<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/spark-Physical-Planning.png"/> </div>
+
+### 4.3 执行
+
+在选择一个物理计划后，Spark运行其RDDs代码，并在运行时执行进一步的优化，生成本地Java字节码，最后将运行结果返回给用户。 
+
+
+
+## 参考资料
+
+1.  Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02 
+2. [Spark SQL, DataFrames and Datasets Guide](https://spark.apache.org/docs/latest/sql-programming-guide.html)
+3. [且谈 Apache Spark 的 API 三剑客：RDD、DataFrame 和 Dataset(译文)](https://www.infoq.cn/article/three-apache-spark-apis-rdds-dataframes-and-datasets)
+4. [A Tale of Three Apache Spark APIs: RDDs vs DataFrames and Datasets(原文)](https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html)