diff --git a/code/spark/spark-core/pom.xml b/code/spark/spark-core/pom.xml index c685072..e8128f4 100644 --- a/code/spark/spark-core/pom.xml +++ b/code/spark/spark-core/pom.xml @@ -10,6 +10,7 @@ 2.12 + 2.4.0 @@ -31,8 +32,14 @@ org.apache.spark spark-core_${scala.version} - 2.4.0 + ${spark.version} + + org.apache.spark + spark-sql_${scala.version} + ${spark.version} + + junit diff --git a/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlTest.scala b/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlTest.scala new file mode 100644 index 0000000..1907cf2 --- /dev/null +++ b/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) + +} diff --git a/notes/SparkSQL-Datasets&DataFrames.md b/notes/SparkSQL-Datasets&DataFrames.md new file mode 100644 index 0000000..58931b4 --- /dev/null +++ b/notes/SparkSQL-Datasets&DataFrames.md @@ -0,0 +1,158 @@ +# Spark SQL DataFrames + + + +## 一、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支持优化器，列式存储和代码生成等特性，以提高查询效率； ++ 支持扩展并能保证容错。 + +
+ +## 二、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\。 + +| 语言 | 主要抽象 | +| ------ | -------------------------------------------- | +| Scala | Dataset[T] & DataFrame (Dataset[Row] 的别名) | +| Java | Dataset[T] | +| Python | DataFrame | +| R | DataFrame | + +### 2.2 DataFrame 对比 RDDs + +DataFrame和RDDs最主要的区别在于一个面向的是结构化数据，一个面向的是非结构化数据，它们内部的数据结构如下： + +
+ +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，后文会对两者做出解释。 + + + +
+ +### 2.4 静态类型与运行时类型安全 + +静态类型(Static-typing)与运行时类型安全(runtime type-safety) 主要表现如下: + +在实际使用中，如果你用的是Spark SQL的查询语句，则直到运行时你才会发现有语法错误，而如果你用的是DataFrame和 Dataset，则在编译时就可以发现错误(这节省了开发时间和整体代价)。DataFrame和Dataset主要区别在于： + +在DataFrame中，当你调用了API之外的函数，编译器就会报错，但如果你使用了一个不存在的字段名字，编译器依然无法发现。而在Dataset中，因为其API都是用lambda函数和JVM类型对象表示的，所有不匹配的类型参数在编译时就会被发现。 + +以上这些最终都被解释成关于类型安全图谱，对应开发中的语法和分析错误。在图谱中，Dataset最严格，但对于开发者来说效率最高。 + +
+ +上面的描述可能并没有那么直观，下面的给出一个IDEA中代码编译的示例： + +
+ +这里一个可能的疑惑是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，并对外提供结构化的访问接口。 + +
+ + + +## 四、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)，优化器是一组规则的集合，用于优化逻辑计划，通过谓词下推等方式进行优化，最终输出优化后的逻辑执行计划。 + +
+ + + +### 4.2 物理计划(Physical Plan) + +在得到优化后的逻辑计划后，Spark就开始了物理计划过程。 它如何通过生成不同的物理执行策略，并通过成本模型来比较它们，从而选择一个最优的物理计划在集群上面执行的。物理规划的输出结果是一系列的RDDs和转换关系(transformations)。 + +
+ +### 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) diff --git a/pictures/spark-Logical-Planning.png b/pictures/spark-Logical-Planning.png new file mode 100644 index 0000000..22ed6ef Binary files /dev/null and b/pictures/spark-Logical-Planning.png differ diff --git a/pictures/spark-Physical-Planning.png b/pictures/spark-Physical-Planning.png new file mode 100644 index 0000000..afbd0fa Binary files /dev/null and b/pictures/spark-Physical-Planning.png differ diff --git a/pictures/spark-dataFrame+RDDs.png b/pictures/spark-dataFrame+RDDs.png new file mode 100644 index 0000000..e1b9bdf Binary files /dev/null and b/pictures/spark-dataFrame+RDDs.png differ diff --git a/pictures/spark-structure-api.png b/pictures/spark-structure-api.png new file mode 100644 index 0000000..98d7a12 Binary files /dev/null and b/pictures/spark-structure-api.png differ diff --git a/pictures/spark-unifed.png b/pictures/spark-unifed.png new file mode 100644 index 0000000..dc7910f Binary files /dev/null and b/pictures/spark-unifed.png differ diff --git a/pictures/spark-运行安全.png b/pictures/spark-运行安全.png new file mode 100644 index 0000000..600c68b Binary files /dev/null and b/pictures/spark-运行安全.png differ diff --git a/pictures/spark-运行时类型安全.png b/pictures/spark-运行时类型安全.png new file mode 100644 index 0000000..0966589 Binary files /dev/null and b/pictures/spark-运行时类型安全.png differ diff --git a/pictures/sql-hive-arch.png b/pictures/sql-hive-arch.png new file mode 100644 index 0000000..9ce9a93 Binary files /dev/null and b/pictures/sql-hive-arch.png differ