From 0dd9d8862facc31dc3a3f168bd915c1177d8d10d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BD=97=E7=A5=A5?= <1366971433@qq.com> Date: Tue, 21 May 2019 16:06:35 +0800 Subject: [PATCH] =?UTF-8?q?spark=20SQL=E5=B8=B8=E7=94=A8=E8=81=9A=E5=90=88?= =?UTF-8?q?=E5=87=BD=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 14 +- .../src/main/java/rdd/scala/SparkSqlApp.scala | 93 +++-- notes/SparkSQL-API基本使用.md | 345 ------------------ ... => SparkSQL_Dataset和DataFrame简介.md} | 0 notes/SparkSQL常用聚合函数.md | 321 ++++++++++++++++ notes/SparkSQL数据源支持.md | 14 + notes/{Spark-RDD.md => Spark_RDD.md} | 0 ...理.md => Spark_Streaming与流处理.md} | 0 notes/Spark_Structured_API的基本使用.md | 182 +++++++++ ...=> Spark_Transformation和Action算子.md} | 0 resources/{ => json}/dept.json | 0 resources/{ => json}/emp.json | 0 resources/parquet/dept.parquet/._SUCCESS.crc | Bin 0 -> 8 bytes ...-b2f4-b780aecfa9ad-c000.snappy.parquet.crc | Bin 0 -> 16 bytes resources/parquet/dept.parquet/_SUCCESS | 0 ...4c2f-b2f4-b780aecfa9ad-c000.snappy.parquet | Bin 0 -> 932 bytes resources/parquet/emp.parquet/._SUCCESS.crc | Bin 0 -> 8 bytes ...-b8bc-0a768ae2ad66-c000.snappy.parquet.crc | Bin 0 -> 32 bytes resources/parquet/emp.parquet/_SUCCESS | 0 ...44d4-b8bc-0a768ae2ad66-c000.snappy.parquet | Bin 0 -> 2661 bytes resources/{dept.txt => tsv/dept.tsv} | 0 resources/{emp.txt => tsv/emp.tsv} | 0 resources/txt/dept.txt | 4 + resources/txt/emp.txt | 14 + 24 files changed, 584 insertions(+), 403 deletions(-) delete mode 100644 notes/SparkSQL-API基本使用.md rename notes/{SparkSQL-Dataset&DataFrame.md => SparkSQL_Dataset和DataFrame简介.md} (100%) create mode 100644 notes/SparkSQL常用聚合函数.md create mode 100644 notes/SparkSQL数据源支持.md rename notes/{Spark-RDD.md => Spark_RDD.md} (100%) rename notes/{Spark-Streaming与流处理.md => Spark_Streaming与流处理.md} (100%) create mode 100644 notes/Spark_Structured_API的基本使用.md rename notes/{Spark-Transformation和Action.md => Spark_Transformation和Action算子.md} (100%) rename resources/{ => json}/dept.json (100%) rename resources/{ => json}/emp.json (100%) create mode 100644 resources/parquet/dept.parquet/._SUCCESS.crc create mode 100644 resources/parquet/dept.parquet/.part-00000-9dcc37bc-f4e8-4c2f-b2f4-b780aecfa9ad-c000.snappy.parquet.crc create mode 100644 resources/parquet/dept.parquet/_SUCCESS create mode 100644 resources/parquet/dept.parquet/part-00000-9dcc37bc-f4e8-4c2f-b2f4-b780aecfa9ad-c000.snappy.parquet create mode 100644 resources/parquet/emp.parquet/._SUCCESS.crc create mode 100644 resources/parquet/emp.parquet/.part-00000-d2881f1b-46f7-44d4-b8bc-0a768ae2ad66-c000.snappy.parquet.crc create mode 100644 resources/parquet/emp.parquet/_SUCCESS create mode 100644 resources/parquet/emp.parquet/part-00000-d2881f1b-46f7-44d4-b8bc-0a768ae2ad66-c000.snappy.parquet rename resources/{dept.txt => tsv/dept.tsv} (100%) rename resources/{emp.txt => tsv/emp.tsv} (100%) create mode 100644 resources/txt/dept.txt create mode 100644 resources/txt/emp.txt diff --git a/README.md b/README.md index 8da6808..4d90b51 100644 --- a/README.md +++ b/README.md @@ -74,20 +74,22 @@ 1. [Spark简介](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark简介.md) 2. [Spark开发环境搭建](https://github.com/heibaiying/BigData-Notes/blob/master/notes/installation/Spark开发环境搭建.md) -4. [弹性式数据集RDD](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark-RDD.md) -5. [RDD常用算子详解](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark-Transformation和Action.md) +4. [弹性式数据集RDD](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark_RDD.md) +5. [RDD常用算子详解](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark_Transformation和Action算子.md) 5. [Spark运行模式与作业提交](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark部署模式与作业提交.md) 6. [Spark累加器与广播变量](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark累加器与广播变量.md) **Spark SQL :** -1. [Spark SQL之 DateFrame 和 DataSet](https://github.com/heibaiying/BigData-Notes/blob/master/notes/SparkSQL-Dataset&DataFrame.md) -2. Spark SQL之常用SQL语句 -3. External Data Source +1. [DateFrames 和 DataSets ](https://github.com/heibaiying/BigData-Notes/blob/master/notes/SparkSQL_Dataset和DataFrame简介.md) +2. [Structured API的基本使用](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Structured_API的基本使用.md) +3. 外部数据源 +4. [Spark SQL常用聚合函数](https://github.com/heibaiying/BigData-Notes/blob/master/notes/SparkSQL常用聚合函数.md) +5. 联结操作 **Spark Streaming ：** -1. [Spark Streaming简介](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark-Streaming与流处理.md) +1. [Spark Streaming简介](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Spark_Streaming与流处理.md) 2. DStream常用操作详解 3. Spark Streaming 整合 Flume 4. Spark Streaming 整合 Kafka diff --git a/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlApp.scala b/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlApp.scala index c70af1f..b40b0de 100644 --- a/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlApp.scala +++ b/code/spark/spark-core/src/main/java/rdd/scala/SparkSqlApp.scala @@ -1,65 +1,54 @@ package rdd.scala -import org.apache.spark.sql.expressions.Aggregator -import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions} +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.functions._ -// 1.定义员工类,对于可能存在null值的字段需要使用Option进行包装 -case class Emp(ename: String, comm: scala.Option[Double], deptno: Long, empno: Long, - hiredate: String, job: String, mgr: scala.Option[Long], sal: Double) - -// 2.定义聚合操作的中间输出类型 -case class SumAndCount(var sum: Double, var count: Long) - -/* 3.自定义聚合函数 - * @IN 聚合操作的输入类型 - * @BUF reduction操作输出值的类型 - * @OUT 聚合操作的输出类型 - */ -object MyAverage extends Aggregator[Emp, SumAndCount, Double] { - - - // 4.用于聚合操作的的初始零值 - override def zero: SumAndCount = SumAndCount(0, 0) - - - // 5.同一分区中的reduce操作 - override def reduce(avg: SumAndCount, emp: Emp): SumAndCount = { - avg.sum += emp.sal - avg.count += 1 - avg - } - - // 6.不同分区中的merge操作 - override def merge(avg1: SumAndCount, avg2: SumAndCount): SumAndCount = { - avg1.sum += avg2.sum - avg1.count += avg2.count - avg1 - } - - // 7.定义最终的输出类型 - override def finish(reduction: SumAndCount): Double = reduction.sum / reduction.count - - // 8.中间类型的编码转换 - override def bufferEncoder: Encoder[SumAndCount] = Encoders.product - - // 9.输出类型的编码转换 - override def outputEncoder: Encoder[Double] = Encoders.scalaDouble -} object SparkSqlApp { // 测试方法 def main(args: Array[String]): Unit = { - val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() - import spark.implicits._ - val ds = spark.read.json("file/emp.json").as[Emp] + val spark = SparkSession.builder().appName("aggregations").master("local[2]").getOrCreate() + val empDF = spark.read.json("/usr/file/json/emp.json") + empDF.createOrReplaceTempView("emp") + empDF.show() + + empDF.select(count("ename")).show() + empDF.select(countDistinct("deptno")).show() + empDF.select(approx_count_distinct("ename", 0.1)).show() + empDF.select(first("ename"), last("job")).show() + empDF.select(min("sal"), max("sal")).show() + empDF.select(sum("sal")).show() + empDF.select(sumDistinct("sal")).show() + empDF.select(avg("sal")).show() + + + // 总体方差 均方差 总体标准差 样本标准差 + empDF.select(var_pop("sal"), var_samp("sal"), stddev_pop("sal"), stddev_samp("sal")).show() + + + // 偏度和峰度 + empDF.select(skewness("sal"), kurtosis("sal")).show() + + // 计算两列的 皮尔逊相关系数 样本协方差 总体协方差 + empDF.select(corr("empno", "sal"), covar_samp("empno", "sal"), + covar_pop("empno", "sal")).show() + + empDF.agg(collect_set("job"), collect_list("ename")).show() + + + empDF.groupBy("deptno", "job").count().show() + spark.sql("SELECT deptno, job, count(*) FROM emp GROUP BY deptno, job").show() + + empDF.groupBy("deptno").agg(count("ename").alias("人数"), sum("sal").alias("总工资")).show() + spark.sql("SELECT deptno, count(ename) ,sum(sal) FROM emp GROUP BY deptno").show() + + + empDF.groupBy("deptno").agg("ename"->"count","sal"->"sum").show() + + - // 10.使用内置avg()函数和自定义函数分别进行计算，验证自定义函数是否正确 - val myAvg = ds.select(MyAverage.toColumn.name("average_sal")).first() - val avg = ds.select(functions.avg(ds.col("sal"))).first().get(0) - println("自定义average函数 : " + myAvg) - println("内置的average函数 : " + avg) } } diff --git a/notes/SparkSQL-API基本使用.md b/notes/SparkSQL-API基本使用.md deleted file mode 100644 index dbcd47b..0000000 --- a/notes/SparkSQL-API基本使用.md +++ /dev/null @@ -1,345 +0,0 @@ -# SparkSQL API基本使用 - -## 一、创建DataFrames - -Spark中所有功能的入口点是`SparkSession`，可以使用`SparkSession.builder()`创建。创建后应用程序就可以从现有RDD，Hive表或Spark数据源创建DataFrame。如下所示： - -```scala -val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() -val df = spark.read.json("/usr/file/emp.json") -df.show() - -// 建议在进行spark SQL编程前导入下面的隐式转换，因为DataFrames和dataSets中很多操作都依赖了隐式转换 -import spark.implicits._ -``` - -这里可以启动`spark-shell`进行测试，需要注意的是`spark-shell`启动后会自动创建一个名为`spark`的`SparkSession`，在命令行中可以直接引用即可： - -![spark-sql-shell](D:\BigData-Notes\pictures\spark-sql-shell.png) - -## 二、DataFrames基本操作 - -### 2.1 printSchema - -```scala -// 以树形结构打印dataframe的schema信息 -df.printSchema() -``` - -![spark-scheme](D:\BigData-Notes\pictures\spark-scheme.png) - -### 2.2 使用DataFrame API进行基本查询 - -```scala -// 查询员工姓名及工作 -df.select($"ename", $"job").show() - -// 查询工资大于2000的员工信息 -df.filter($"sal" > 2000).show() - -// 分组统计部门人数 -df.groupBy("deptno").count().show() -``` - -### 2.3 使用SQL进行基本查询 - -```scala -// 首先需要将DataFrame注册为临时视图 -df.createOrReplaceTempView("emp") - -// 查询员工姓名及工作 -spark.sql("SELECT ename,job FROM emp").show() - -// 查询工资大于2000的员工信息 -spark.sql("SELECT * FROM emp where sal > 2000").show() - -// 分组统计部门人数 -spark.sql("SELECT deptno,count(ename) FROM emp group by deptno").show() -``` - -### 2.4 全局临时视图 - -上面使用`createOrReplaceTempView`创建的是会话临时视图，它的生命周期仅限于会话范围，会随会话的结束而结束。 - -你也可以使用`createGlobalTempView`创建全局临时视图，全局临时视图可以在所有会话之间共享，并直到整个Spark应用程序终止才会消失。全局临时视图被定义在内置的`global_temp`数据库下，需要使用限定名称进行引用，如`SELECT * FROM global_temp.view1`。 - -```scala -// 注册为全局临时视图 -df.createGlobalTempView("gemp") - -// 查询员工姓名及工作，使用限定名称进行引用 -spark.sql("SELECT ename,job FROM global_temp.gemp").show() - -// 查询工资大于2000的员工信息，使用限定名称进行引用 -spark.sql("SELECT * FROM global_temp.gemp where sal > 2000").show() - -// 分组统计部门人数，使用限定名称进行引用 -spark.sql("SELECT deptno,count(ename) FROM global_temp.gemp group by deptno").show() -``` - -## 三、创建Datasets - -### 3.1 由外部数据集创建 - -```scala -// 1.需要导入隐式转换 -import spark.implicits._ - -// 2.创建case class,等价于Java Bean -case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, - hiredate: String, job: String, mgr: Long, sal: Double) - -// 3.由外部数据集创建Datasets -val ds = spark.read.json("/usr/file/emp.json").as[Emp] -ds.show() -``` - -### 3.2 由内部数据集创建 - -```scala -// 1.需要导入隐式转换 -import spark.implicits._ - -// 2.创建case class,等价于Java Bean -case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, - hiredate: String, job: String, mgr: Long, sal: Double) - -// 3.由内部数据集创建Datasets -val caseClassDS = Seq(Emp("ALLEN", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0), - Emp("JONES", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0)) - .toDS() -caseClassDS.show() -``` - - - -## 四、DataFrames与Datasets互相转换 - -Spark提供了非常简单的转换方法用于DataFrames与Datasets互相转换，示例如下： - -```shell -# DataFrames转Datasets -scala> df.as[Emp] -res1: org.apache.spark.sql.Dataset[Emp] = [COMM: double, DEPTNO: bigint ... 6 more fields] - -# Datasets转DataFrames -scala> ds.toDF() -res2: org.apache.spark.sql.DataFrame = [COMM: double, DEPTNO: bigint ... 6 more fields] -``` - - - -## 五、RDDs转换为DataFrames\Datasets - -Spark支持两种方式把RDD转换为DataFrames，分别是使用反射推断和指定schema转换。 - -### 5.1 使用反射推断 - -```scala -// 1.导入隐式转换 -import spark.implicits._ - -// 2.创建部门类 -case class Dept(deptno: Long, dname: String, loc: String) - -// 3.创建RDD并转换为dataSet -val rddToDS = spark.sparkContext - .textFile("/usr/file/dept.txt") - .map(_.split("\t")) - .map(line => Dept(line(0).trim.toLong, line(1), line(2))) - .toDS() // 如果调用toDF()则转换为dataFrame -``` - -### 5.2 以编程方式指定Schema - -```scala -import org.apache.spark.sql.Row -import org.apache.spark.sql.types._ - - -// 1.定义每个列的列类型 -val fields = Array(StructField("deptno", LongType, nullable = true), - StructField("dname", StringType, nullable = true), - StructField("loc", StringType, nullable = true)) - -// 2.创建schema -val schema = StructType(fields) - -// 3.创建RDD -val deptRDD = spark.sparkContext.textFile("/usr/file/dept.txt") -val rowRDD = deptRDD.map(_.split("\t")).map(line => Row(line(0).toLong, line(1), line(2))) - - -// 4.将RDD转换为dataFrame -val deptDF = spark.createDataFrame(rowRDD, schema) -deptDF.show() -``` - -## 六、使用自定义聚合函数 - -Scala提供了两种自定义聚合函数的方法，分别如下： - -+ 有类型的自定义聚合函数，主要适用于DataSets； -+ 无类型的自定义聚合函数，主要适用于DataFrames。 - -以下分别使用两种方式来自定义一个求平均值的聚合函数，这里以计算员工平均工资为例。两种自定义方式分别如下： - -### 6.1 有类型的自定义函数 - -```scala -import org.apache.spark.sql.expressions.Aggregator -import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions} - -// 1.定义员工类,对于可能存在null值的字段需要使用Option进行包装 -case class Emp(ename: String, comm: scala.Option[Double], deptno: Long, empno: Long, - hiredate: String, job: String, mgr: scala.Option[Long], sal: Double) - -// 2.定义聚合操作的中间输出类型 -case class SumAndCount(var sum: Double, var count: Long) - -/* 3.自定义聚合函数 - * @IN 聚合操作的输入类型 - * @BUF reduction操作输出值的类型 - * @OUT 聚合操作的输出类型 - */ -object MyAverage extends Aggregator[Emp, SumAndCount, Double] { - - // 4.用于聚合操作的的初始零值 - override def zero: SumAndCount = SumAndCount(0, 0) - - // 5.同一分区中的reduce操作 - override def reduce(avg: SumAndCount, emp: Emp): SumAndCount = { - avg.sum += emp.sal - avg.count += 1 - avg - } - - // 6.不同分区中的merge操作 - override def merge(avg1: SumAndCount, avg2: SumAndCount): SumAndCount = { - avg1.sum += avg2.sum - avg1.count += avg2.count - avg1 - } - - // 7.定义最终的输出类型 - override def finish(reduction: SumAndCount): Double = reduction.sum / reduction.count - - // 8.中间类型的编码转换 - override def bufferEncoder: Encoder[SumAndCount] = Encoders.product - - // 9.输出类型的编码转换 - override def outputEncoder: Encoder[Double] = Encoders.scalaDouble -} - -object SparkSqlApp { - - // 测试方法 - def main(args: Array[String]): Unit = { - - val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() - import spark.implicits._ - val ds = spark.read.json("file/emp.json").as[Emp] - - // 10.使用内置avg()函数和自定义函数分别进行计算，验证自定义函数是否正确 - val myAvg = ds.select(MyAverage.toColumn.name("average_sal")).first() - val avg = ds.select(functions.avg(ds.col("sal"))).first().get(0) - - println("自定义average函数 : " + myAvg) - println("内置的average函数 : " + avg) - } -} -``` - -自定义聚合函数需要实现的方法比较多，这里以绘图的方式来演示其执行流程，以及每个方法的作用： - -![spark-sql-自定义函数](D:\BigData-Notes\pictures\spark-sql-自定义函数.png) - - - -关于`zero`,`reduce`,`merge`,`finish`方法的作用在上图都有说明，这里解释一下中间类型和输出类型的编码转换，这个写法比较固定，基本上就是两种情况： - -+ 自定义类型case class或者元组就使用`Encoders.product`方法； -+ 基本类型就使用其对应名称的方法，如`scalaByte `，`scalaFloat`，`scalaShort`等。 - -```scala -override def bufferEncoder: Encoder[SumAndCount] = Encoders.product -override def outputEncoder: Encoder[Double] = Encoders.scalaDouble -``` - - - -### 6.2 无类型的自定义聚合函数 - -理解了有类型的自定义聚合函数后，无类型的定义方式也基本相同，代码如下： - -```scala -import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} -import org.apache.spark.sql.types._ -import org.apache.spark.sql.{Row, SparkSession} - -object MyAverage extends UserDefinedAggregateFunction { - // 1.聚合操作输入参数的类型,字段名称可以自定义 - def inputSchema: StructType = StructType(StructField("MyInputColumn", LongType) :: Nil) - - // 2.聚合操作中间值的类型,字段名称可以自定义 - def bufferSchema: StructType = { - StructType(StructField("sum", LongType) :: StructField("MyCount", LongType) :: Nil) - } - - // 3.聚合操作输出参数的类型 - def dataType: DataType = DoubleType - - // 4.此函数是否始终在相同输入上返回相同的输出,通常为true - def deterministic: Boolean = true - - // 5.定义零值 - def initialize(buffer: MutableAggregationBuffer): Unit = { - buffer(0) = 0L - buffer(1) = 0L - } - - // 6.同一分区中的reduce操作 - def update(buffer: MutableAggregationBuffer, input: Row): Unit = { - if (!input.isNullAt(0)) { - buffer(0) = buffer.getLong(0) + input.getLong(0) - buffer(1) = buffer.getLong(1) + 1 - } - } - - // 7.不同分区中的merge操作 - def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { - buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0) - buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1) - } - - // 8.计算最终的输出值 - def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1) -} - -object SparkSqlApp { - - // 测试方法 - def main(args: Array[String]): Unit = { - - val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() - // 9.注册自定义的聚合函数 - spark.udf.register("myAverage", MyAverage) - - val df = spark.read.json("file/emp.json") - df.createOrReplaceTempView("emp") - - // 10.使用自定义函数和内置函数分别进行计算 - val myAvg = spark.sql("SELECT myAverage(sal) as avg_sal FROM emp").first() - val avg = spark.sql("SELECT avg(sal) as avg_sal FROM emp").first() - - println("自定义average函数 : " + myAvg) - println("内置的average函数 : " + avg) - } -} -``` - - - -## 参考资料 - -[Spark SQL, DataFrames and Datasets Guide > Getting Started](https://spark.apache.org/docs/latest/sql-getting-started.html) \ No newline at end of file diff --git a/notes/SparkSQL-Dataset&DataFrame.md b/notes/SparkSQL_Dataset和DataFrame简介.md similarity index 100% rename from notes/SparkSQL-Dataset&DataFrame.md rename to notes/SparkSQL_Dataset和DataFrame简介.md diff --git a/notes/SparkSQL常用聚合函数.md b/notes/SparkSQL常用聚合函数.md new file mode 100644 index 0000000..b214010 --- /dev/null +++ b/notes/SparkSQL常用聚合函数.md @@ -0,0 +1,321 @@ +# 聚合函数Aggregations + +## 一、简单聚合 + +### 1.1 数据准备 + +```scala +// 需要导入spark sql内置的函数包 +import org.apache.spark.sql.functions._ + +val spark = SparkSession.builder().appName("aggregations").master("local[2]").getOrCreate() +val empDF = spark.read.json("/usr/file/json/emp.json") +// 注册为临时视图，用于后面演示SQL查询 +empDF.createOrReplaceTempView("emp") +empDF.show() +``` + +> 注：emp.json可以在本仓库的resources目录进行下载。 + +### 1.2 count + +```scala +// 计算员工人数 +empDF.select(count("ename")).show() +``` + +### 1.3 countDistinct + +```scala +// 计算姓名不重复的员工人数 +empDF.select(countDistinct("deptno")).show() +``` + +### 1.4 approx_count_distinct + +通常在使用大型数据集时，你可能关注的只是近似值而不是准确值，这时可以使用approx_count_distinct函数，并可以使用第二个参数指定最大允许误差。 + +```scala +empDF.select(approx_count_distinct ("ename",0.1)).show() +``` + +### 1.5 first & last + +获取DataFrame中指定列的第一个值或者最后一个值。 + +```scala +empDF.select(first("ename"),last("job")).show() +``` + +### 1.6 min & max + +获取DataFrame中指定列的最小值或者最大值。 + +```scala +empDF.select(min("sal"),max("sal")).show() +``` + +### 1.7 sum & sumDistinct + +求和以及求指定列所有不相同的值的和。 + +```scala +empDF.select(sum("sal")).show() +empDF.select(sumDistinct("sal")).show() +``` + +### 1.8 avg + +内置的求平均数的函数。 + +```scala +empDF.select(avg("sal")).show() +``` + +### 1.9 数学函数 + +Spark SQL中还支持多种数学聚合函数，用于通常的数学计算，以下是一些常用的例子： + +```scala +// 1.计算总体方差、均方差、总体标准差、样本标准差 +empDF.select(var_pop("sal"), var_samp("sal"), stddev_pop("sal"), stddev_samp("sal")).show() + +// 2.计算偏度和峰度 +empDF.select(skewness("sal"), kurtosis("sal")).show() + +// 3. 计算两列的皮尔逊相关系数、样本协方差、总体协方差。(这里只是演示，员工编号和薪资两列实际上并没有什么关联关系) +empDF.select(corr("empno", "sal"), covar_samp("empno", "sal"),covar_pop("empno", "sal")).show() +``` + +### 1.10 聚合数据到集合 + +```scala +scala> empDF.agg(collect_set("job"), collect_list("ename")).show() + +输出： ++--------------------+--------------------+ +| collect_set(job)| collect_list(ename)| ++--------------------+--------------------+ +|[MANAGER, SALESMA...|[SMITH, ALLEN, WA...| ++--------------------+--------------------+ +``` + + + +## 二、分组聚合 + +### 2.1 简单分组 + +```scala +empDF.groupBy("deptno", "job").count().show() +//等价SQL +spark.sql("SELECT deptno, job, count(*) FROM emp GROUP BY deptno, job").show() + +输出： ++------+---------+-----+ +|deptno| job|count| ++------+---------+-----+ +| 10|PRESIDENT| 1| +| 30| CLERK| 1| +| 10| MANAGER| 1| +| 30| MANAGER| 1| +| 20| CLERK| 2| +| 30| SALESMAN| 4| +| 20| ANALYST| 2| +| 10| CLERK| 1| +| 20| MANAGER| 1| ++------+---------+-----+ +``` + +### 2.2 分组聚合 + +```scala +empDF.groupBy("deptno").agg(count("ename").alias("人数"), sum("sal").alias("总工资")).show() +// 等价语法 +empDF.groupBy("deptno").agg("ename"->"count","sal"->"sum").show() +// 等价SQL +spark.sql("SELECT deptno, count(ename) ,sum(sal) FROM emp GROUP BY deptno").show() + +输出： ++------+----+------+ +|deptno|人数|总工资| ++------+----+------+ +| 10| 3|8750.0| +| 30| 6|9400.0| +| 20| 5|9375.0| ++------+----+------+ +``` + + + +## 三、自定义聚合函数 + +Scala提供了两种自定义聚合函数的方法，分别如下： + +- 有类型的自定义聚合函数，主要适用于DataSets； +- 无类型的自定义聚合函数，主要适用于DataFrames。 + +以下分别使用两种方式来自定义一个求平均值的聚合函数，这里以计算员工平均工资为例。两种自定义方式分别如下： + +### 3.1 有类型的自定义函数 + +```scala +import org.apache.spark.sql.expressions.Aggregator +import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions} + +// 1.定义员工类,对于可能存在null值的字段需要使用Option进行包装 +case class Emp(ename: String, comm: scala.Option[Double], deptno: Long, empno: Long, + hiredate: String, job: String, mgr: scala.Option[Long], sal: Double) + +// 2.定义聚合操作的中间输出类型 +case class SumAndCount(var sum: Double, var count: Long) + +/* 3.自定义聚合函数 + * @IN 聚合操作的输入类型 + * @BUF reduction操作输出值的类型 + * @OUT 聚合操作的输出类型 + */ +object MyAverage extends Aggregator[Emp, SumAndCount, Double] { + + // 4.用于聚合操作的的初始零值 + override def zero: SumAndCount = SumAndCount(0, 0) + + // 5.同一分区中的reduce操作 + override def reduce(avg: SumAndCount, emp: Emp): SumAndCount = { + avg.sum += emp.sal + avg.count += 1 + avg + } + + // 6.不同分区中的merge操作 + override def merge(avg1: SumAndCount, avg2: SumAndCount): SumAndCount = { + avg1.sum += avg2.sum + avg1.count += avg2.count + avg1 + } + + // 7.定义最终的输出类型 + override def finish(reduction: SumAndCount): Double = reduction.sum / reduction.count + + // 8.中间类型的编码转换 + override def bufferEncoder: Encoder[SumAndCount] = Encoders.product + + // 9.输出类型的编码转换 + override def outputEncoder: Encoder[Double] = Encoders.scalaDouble +} + +object SparkSqlApp { + + // 测试方法 + def main(args: Array[String]): Unit = { + + val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() + import spark.implicits._ + val ds = spark.read.json("file/emp.json").as[Emp] + + // 10.使用内置avg()函数和自定义函数分别进行计算，验证自定义函数是否正确 + val myAvg = ds.select(MyAverage.toColumn.name("average_sal")).first() + val avg = ds.select(functions.avg(ds.col("sal"))).first().get(0) + + println("自定义average函数 : " + myAvg) + println("内置的average函数 : " + avg) + } +} +``` + +自定义聚合函数需要实现的方法比较多，这里以绘图的方式来演示其执行流程，以及每个方法的作用： + +![spark-sql-自定义函数](D:\BigData-Notes\pictures\spark-sql-自定义函数.png) + + + +关于`zero`,`reduce`,`merge`,`finish`方法的作用在上图都有说明，这里解释一下中间类型和输出类型的编码转换，这个写法比较固定，基本上就是两种情况： + +- 自定义类型case class或者元组就使用`Encoders.product`方法； +- 基本类型就使用其对应名称的方法，如`scalaByte `，`scalaFloat`，`scalaShort`等。 + +```scala +override def bufferEncoder: Encoder[SumAndCount] = Encoders.product +override def outputEncoder: Encoder[Double] = Encoders.scalaDouble +``` + + + +### 3.2 无类型的自定义聚合函数 + +理解了有类型的自定义聚合函数后，无类型的定义方式也基本相同，代码如下： + +```scala +import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} +import org.apache.spark.sql.types._ +import org.apache.spark.sql.{Row, SparkSession} + +object MyAverage extends UserDefinedAggregateFunction { + // 1.聚合操作输入参数的类型,字段名称可以自定义 + def inputSchema: StructType = StructType(StructField("MyInputColumn", LongType) :: Nil) + + // 2.聚合操作中间值的类型,字段名称可以自定义 + def bufferSchema: StructType = { + StructType(StructField("sum", LongType) :: StructField("MyCount", LongType) :: Nil) + } + + // 3.聚合操作输出参数的类型 + def dataType: DataType = DoubleType + + // 4.此函数是否始终在相同输入上返回相同的输出,通常为true + def deterministic: Boolean = true + + // 5.定义零值 + def initialize(buffer: MutableAggregationBuffer): Unit = { + buffer(0) = 0L + buffer(1) = 0L + } + + // 6.同一分区中的reduce操作 + def update(buffer: MutableAggregationBuffer, input: Row): Unit = { + if (!input.isNullAt(0)) { + buffer(0) = buffer.getLong(0) + input.getLong(0) + buffer(1) = buffer.getLong(1) + 1 + } + } + + // 7.不同分区中的merge操作 + def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { + buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0) + buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1) + } + + // 8.计算最终的输出值 + def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1) +} + +object SparkSqlApp { + + // 测试方法 + def main(args: Array[String]): Unit = { + + val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() + // 9.注册自定义的聚合函数 + spark.udf.register("myAverage", MyAverage) + + val df = spark.read.json("file/emp.json") + df.createOrReplaceTempView("emp") + + // 10.使用自定义函数和内置函数分别进行计算 + val myAvg = spark.sql("SELECT myAverage(sal) as avg_sal FROM emp").first() + val avg = spark.sql("SELECT avg(sal) as avg_sal FROM emp").first() + + println("自定义average函数 : " + myAvg) + println("内置的average函数 : " + avg) + } +} +``` + + + +## 参考资料 + +1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02 + + + diff --git a/notes/SparkSQL数据源支持.md b/notes/SparkSQL数据源支持.md new file mode 100644 index 0000000..fff2e24 --- /dev/null +++ b/notes/SparkSQL数据源支持.md @@ -0,0 +1,14 @@ +1.1 Json + +```scala +val empDF = spark.read.json("/usr/file/json/emp.json") +empDF.show() +``` + +1.2 + +```scala +val parquetFileDF = spark.read.parquet("/usr/file/parquet/emp.parquet") +parquetFileDF.show() +``` + diff --git a/notes/Spark-RDD.md b/notes/Spark_RDD.md similarity index 100% rename from notes/Spark-RDD.md rename to notes/Spark_RDD.md diff --git a/notes/Spark-Streaming与流处理.md b/notes/Spark_Streaming与流处理.md similarity index 100% rename from notes/Spark-Streaming与流处理.md rename to notes/Spark_Streaming与流处理.md diff --git a/notes/Spark_Structured_API的基本使用.md b/notes/Spark_Structured_API的基本使用.md new file mode 100644 index 0000000..189eb73 --- /dev/null +++ b/notes/Spark_Structured_API的基本使用.md @@ -0,0 +1,182 @@ +# Structured API基本使用 + +## 一、创建DataFrames + +Spark中所有功能的入口点是`SparkSession`，可以使用`SparkSession.builder()`创建。创建后应用程序就可以从现有RDD，Hive表或Spark数据源创建DataFrame。如下所示： + +```scala +val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() +val df = spark.read.json("/usr/file/emp.json") +df.show() + +// 建议在进行spark SQL编程前导入下面的隐式转换，因为DataFrames和dataSets中很多操作都依赖了隐式转换 +import spark.implicits._ +``` + +这里可以启动`spark-shell`进行测试，需要注意的是`spark-shell`启动后会自动创建一个名为`spark`的`SparkSession`，在命令行中可以直接引用即可： + +![spark-sql-shell](D:\BigData-Notes\pictures\spark-sql-shell.png) + +## 二、DataFrames基本操作 + +### 2.1 printSchema + +```scala +// 以树形结构打印dataframe的schema信息 +df.printSchema() +``` + +![spark-scheme](D:\BigData-Notes\pictures\spark-scheme.png) + +### 2.2 使用DataFrame API进行基本查询 + +```scala +// 查询员工姓名及工作 +df.select($"ename", $"job").show() + +// 查询工资大于2000的员工信息 +df.filter($"sal" > 2000).show() + +// 分组统计部门人数 +df.groupBy("deptno").count().show() +``` + +### 2.3 使用SQL进行基本查询 + +```scala +// 首先需要将DataFrame注册为临时视图 +df.createOrReplaceTempView("emp") + +// 查询员工姓名及工作 +spark.sql("SELECT ename,job FROM emp").show() + +// 查询工资大于2000的员工信息 +spark.sql("SELECT * FROM emp where sal > 2000").show() + +// 分组统计部门人数 +spark.sql("SELECT deptno,count(ename) FROM emp group by deptno").show() +``` + +### 2.4 全局临时视图 + +上面使用`createOrReplaceTempView`创建的是会话临时视图，它的生命周期仅限于会话范围，会随会话的结束而结束。 + +你也可以使用`createGlobalTempView`创建全局临时视图，全局临时视图可以在所有会话之间共享，并直到整个Spark应用程序终止才会消失。全局临时视图被定义在内置的`global_temp`数据库下，需要使用限定名称进行引用，如`SELECT * FROM global_temp.view1`。 + +```scala +// 注册为全局临时视图 +df.createGlobalTempView("gemp") + +// 查询员工姓名及工作，使用限定名称进行引用 +spark.sql("SELECT ename,job FROM global_temp.gemp").show() + +// 查询工资大于2000的员工信息，使用限定名称进行引用 +spark.sql("SELECT * FROM global_temp.gemp where sal > 2000").show() + +// 分组统计部门人数，使用限定名称进行引用 +spark.sql("SELECT deptno,count(ename) FROM global_temp.gemp group by deptno").show() +``` + +## 三、创建Datasets + +### 3.1 由外部数据集创建 + +```scala +// 1.需要导入隐式转换 +import spark.implicits._ + +// 2.创建case class,等价于Java Bean +case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, + hiredate: String, job: String, mgr: Long, sal: Double) + +// 3.由外部数据集创建Datasets +val ds = spark.read.json("/usr/file/emp.json").as[Emp] +ds.show() +``` + +### 3.2 由内部数据集创建 + +```scala +// 1.需要导入隐式转换 +import spark.implicits._ + +// 2.创建case class,等价于Java Bean +case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, + hiredate: String, job: String, mgr: Long, sal: Double) + +// 3.由内部数据集创建Datasets +val caseClassDS = Seq(Emp("ALLEN", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0), + Emp("JONES", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0)) + .toDS() +caseClassDS.show() +``` + + + +## 四、DataFrames与Datasets互相转换 + +Spark提供了非常简单的转换方法用于DataFrames与Datasets互相转换，示例如下： + +```shell +# DataFrames转Datasets +scala> df.as[Emp] +res1: org.apache.spark.sql.Dataset[Emp] = [COMM: double, DEPTNO: bigint ... 6 more fields] + +# Datasets转DataFrames +scala> ds.toDF() +res2: org.apache.spark.sql.DataFrame = [COMM: double, DEPTNO: bigint ... 6 more fields] +``` + + + +## 五、RDDs转换为DataFrames\Datasets + +Spark支持两种方式把RDD转换为DataFrames，分别是使用反射推断和指定schema转换。 + +### 5.1 使用反射推断 + +```scala +// 1.导入隐式转换 +import spark.implicits._ + +// 2.创建部门类 +case class Dept(deptno: Long, dname: String, loc: String) + +// 3.创建RDD并转换为dataSet +val rddToDS = spark.sparkContext + .textFile("/usr/file/dept.txt") + .map(_.split("\t")) + .map(line => Dept(line(0).trim.toLong, line(1), line(2))) + .toDS() // 如果调用toDF()则转换为dataFrame +``` + +### 5.2 以编程方式指定Schema + +```scala +import org.apache.spark.sql.Row +import org.apache.spark.sql.types._ + + +// 1.定义每个列的列类型 +val fields = Array(StructField("deptno", LongType, nullable = true), + StructField("dname", StringType, nullable = true), + StructField("loc", StringType, nullable = true)) + +// 2.创建schema +val schema = StructType(fields) + +// 3.创建RDD +val deptRDD = spark.sparkContext.textFile("/usr/file/dept.txt") +val rowRDD = deptRDD.map(_.split("\t")).map(line => Row(line(0).toLong, line(1), line(2))) + + +// 4.将RDD转换为dataFrame +val deptDF = spark.createDataFrame(rowRDD, schema) +deptDF.show() +``` + + + +## 参考资料 + +[Spark SQL, DataFrames and Datasets Guide > Getting Started](https://spark.apache.org/docs/latest/sql-getting-started.html) \ No newline at end of file diff --git a/notes/Spark-Transformation和Action.md b/notes/Spark_Transformation和Action算子.md similarity index 100% rename from notes/Spark-Transformation和Action.md rename to notes/Spark_Transformation和Action算子.md diff --git a/resources/dept.json b/resources/json/dept.json similarity index 100% rename from resources/dept.json rename to resources/json/dept.json diff --git a/resources/emp.json b/resources/json/emp.json similarity index 100% rename from resources/emp.json rename to resources/json/emp.json diff --git a/resources/parquet/dept.parquet/._SUCCESS.crc b/resources/parquet/dept.parquet/._SUCCESS.crc new file mode 100644 index 0000000000000000000000000000000000000000..3b7b044936a890cd8d651d349a752d819d71d22c GIT binary patch literal 8 PcmYc;N@ieSU}69O2$TUk literal 0 HcmV?d00001 diff --git a/resources/parquet/dept.parquet/.part-00000-9dcc37bc-f4e8-4c2f-b2f4-b780aecfa9ad-c000.snappy.parquet.crc b/resources/parquet/dept.parquet/.part-00000-9dcc37bc-f4e8-4c2f-b2f4-b780aecfa9ad-c000.snappy.parquet.crc new file mode 100644 index 0000000000000000000000000000000000000000..78cadc30090f427f16c7b0730bd0701da8cb0476 GIT binary patch literal 16 XcmYc;N@ieSU}7ltFC=+AorXwT}#Ry^?aE0W1j2Bg95xZ3-W4V-4N*UWqY9e1E$t}gW z>pV^uS%=a^9-;JcdSux+rP){rac2Jc{xkCr>S=wIAYzhskzgoTf<+z|Q2g*Xv#mH( zcw`pDyqJJ*@<|3AGwA8@(F^o!i|kCmbw>QXeGL`Tj%FGTpV4&PerpZ7);TcsKRYLc z_5uc_Z#ah5*E?ybH)OOcyJz&ZLD#k%XfPxf|1`js@hqR`4?k)(twW|6RCNP@%MJ;U8GdE7c6Ye9sAA`%@Uf6s^{yS0tG zZ(Q#f?iT}M>+y^qI@~1p4$+P`{11k^W%4)4i2pyr2RL_|;xH0K{&enN%5*^?9C?$I z99(-c1a`f2L)o8ym8Wh<$26q;Z)PQh-xjXal!DM-jzXy@T})hW97xR%D``fjkrd|& z(srpgpIyRew)8wY^k4;`Zc*B~O3l@JUEI^rMeJedGdTo@91ev)t`Yxgd*0_$KljD; zgUNiBQ)N|=m0V%CoOt7$QdViLR;?(?nOhz1M_#ZhCx3x|V0JrlEQ2+n{ literal 0 HcmV?d00001 diff --git a/resources/parquet/emp.parquet/_SUCCESS b/resources/parquet/emp.parquet/_SUCCESS new file mode 100644 index 0000000..e69de29 diff --git a/resources/parquet/emp.parquet/part-00000-d2881f1b-46f7-44d4-b8bc-0a768ae2ad66-c000.snappy.parquet b/resources/parquet/emp.parquet/part-00000-d2881f1b-46f7-44d4-b8bc-0a768ae2ad66-c000.snappy.parquet new file mode 100644 index 0000000000000000000000000000000000000000..0cfd4880ba99e882519d7e4b2b483a3fd27ba2a0 GIT binary patch literal 2661 zcmbVOPfR0K7=Ld%15+q&UHYa|9X3q^l}(uDO({@FLpsnd|E7hs6|-@TJ3v8V`Liux z3=%KIuyG|(SKT!x9yG3r7n8l1J(0~G{BzI~2eR3C@?bo9FdkOp_hzOPmnI7|oj31& z-~0W(KkxfqJEi9|f=HH(b`b?84MbFA5wXv&A0?d`EF@oN9VsYYP+Twy5{EDr_sU7! zD5B)sZm|>Kps+rKCzM1CAm?>JH2A&b>L66zF>OM$@C z#lfgxPVjYH5+RH~F!>Z8jp$zoCyPHQwnavETnf64piKrF1*n z0xa^GvBD^JQ{6DqSx6URd-1dJ@yTpD-vH~3o-2%HajSR8&?nN^?HxAs+yo%zQ;Ox5q9`=28~Z7}&@`zV~_ZF=vqH*ztVBrCNbQ z>yJ}ai>g{*kkSFc)#8^ml}6Plo%SOX9rq(KrqZk*dHdC9JSbe|2WZzeo6LkqoVBnJf;VAjl^49x!p5z8KOqQpm6Q$A!~%`K#nL?t8JJ*|o!5Ai&R#(OwacLgpY2>&?7wr;Ee@2xUcHH) zdh;FidQVV@$4P~qR*Uu*PI|zF_k_fgm)KG4Hl)$QLd1uMV8K6VkoH~SZIQuG-pO^M zK4`9)V8?2)|AnQ;IDF8|*y}W89h&`?+S#y*uxM^_(l!^~Z-LaExY_D(ADnZF39`Uy z%;Nr=lkRijgLBnlI!T`#=a&8?l7a$?d~v0`5HVLx=#+@nG?B`+QUto^+In#41Bxs~;q zQW4h6>!p&(rUL*ic7>{}p@GeI+&y(-Sm z07ZCqFqB>me7dYbW!0c3JFr)_kUZ`biMmt{ZWPOv#g*lt7SU*g20Lfg7fW+N+O3(r zy*)8X`-|N(@hFW_szpsL9_x