重温拉链表|Spark SQL|
拉链表
先放到增量表中，然后进行关联到一张临时表中，在插入到新表中 1、 判断失效值，
2、 判断有效值，
3、通过UNION ALL进行联合
DROP TABLE IF EXISTS dw_orders_his_tmp;
CREATE TABLE dw_orders_his_tmp AS
SELECT orderid,
createtime,
modifiedtime,
status,
dw_start_date,
dw_end_date
FROM (
//判断失效值
SELECT a.orderid,
a.createtime,
a.modifiedtime,
a.status,
a.dw_start_date,
CASE WHEN b.orderid IS NOT NULL AND a.dw_end_date > '2016-08-21' THEN '2016-08-21' ELSE a.dw_end_date END AS dw_end_date
FROM dw_orders_his a
left outer join (SELECT * FROM ods_orders_inc WHERE day = '2016-08-21') b
ON (a.orderid = b.orderid)
UNION ALL
//判断有效值
SELECT orderid,
createtime,
modifiedtime,
status,
modifiedtime AS dw_start_date,
'9999-12-31' AS dw_end_date
FROM ods_orders_inc
WHERE day = '2016-08-21'
) x
ORDER BY orderid,dw_start_date;
INSERT overwrite TABLE dw_orders_his
SELECT * FROM dw_orders_his_tmp;
Spark SQL
DateFrame
什么是DateFrame
DataFrame 是一种以 RDD 为基础的分布式数据集，类似于传统数据库中的二维表格。
DataFrame 与 RDD的主要区别在于，前者带有 schema 元信息，即 DataFrame所表示的二维表数据集的每一列都带有名称和类型。
RDD ==> DateFrame
在 IDEA 中开发程序时，如果需要 RDD 与 DF 或者 DS 之间互相操作，那么需要引入 importspark.implicits._ 这里的 spark 不是 Scala 中的包名，而是创建的 sparkSession
对象的变量名称，所以必 须先创建 SparkSession 对象再导入。这里的 spark 对象不能使用 var 声明，因为 Scala 只支持val 修饰的对象的引入。
//实际开发中，一般通过样例类将 RDD 转换为 DataFrame
scala> case class User(name:String, age:Int)
defined class User
scala> sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1,
t._2)).toDF.show
DateFrame ==> RDD
//DataFrame 其实就是对 RDD 的封装，所以可以直接获取内部的 RDD
scala> val df = sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1,
t._2)).toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: int]
scala> val rdd = df.rdd
rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[46]
at rdd at <console>:25
scala> val array = rdd.collect
array: Array[org.apache.spark.sql.Row] = Array([zhangsan,30], [lisi,40])
//注意：此时得到的 RDD 存储类型为 Row
scala> array(0)
res28: org.apache.spark.sql.Row = [zhangsan,30]
scala> array(0)(0)
res29: Any = zhangsan
scala> array(0).getAs[String]("name")
res30: String = zhangsan
DateSet
DataSet 是具有强类型的数据集合，需要提供对应的类型信息。
RDD ? DateSet
SparkSQL 能够自动将包含有 case 类的 RDD 转换成 DataSet，case 类定义了 table 的结构，case
类属性通过反射变成了表的列名。Case 类可以包含诸如 Seq 或者 Array 等复杂的结构。
scala> case class User(name:String, age:Int)
defined class User
scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1,
t._2)).toDS
res11: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
DateSet ==> RDD
DataSet 其实也是对 RDD 的封装，所以可以直接获取内部的 RDD
scala> case class User(name:String, age:Int)
defined class User
scala> sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1,
t._2)).toDS
res11: org.apache.spark.sql.Dataset[User] = [name: string, age: int]
scala> val rdd = res11.rdd
rdd: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[51] at rdd at
<console>:25
scala> rdd.collect
res12: Array[User] = Array(User(zhangsan,30), User(lisi,49))
RDD、DataFrame、DataSet 三者的关系
相同点
? RDD、DataFrame、DataSet 全都是 spark 平台下的分布式弹性数据集，为处理超大型数 据提供便利; ?
三者都有惰性机制，在进行创建、转换，如 map 方法时，不会立即执行，只有在遇到 Action 如 foreach
时，三者才会开始遍历运算; ? 三者有许多共同的函数，如 filter，排序等; ? 在对 DataFrame 和 Dataset
进行操作许多操作都需要这个包:import spark.implicits._（在 创建好 SparkSession 对象后尽量直接导入）
? 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会 内存溢出 ? 三者都有 partition 的概念
? DataFrame 和 DataSet 均可使用模式匹配获取各个字段的值和类型
三者的区别
-
RDD
? RDD 一般和 spark mllib 同时使用
? RDD 不支持 sparksql 操作 -
DataFrame
? 与 RDD 和 Dataset 不同，DataFrame 每一行的类型固定为 Row，每一列的值没法直
接访问，只有通过解析才能获取各个字段的值
? DataFrame 与 DataSet 一般不与 spark mllib 同时使用
? DataFrame 与 DataSet 均支持 SparkSQL 的操作，比如 select，groupby 之类，还能
注册临时表/视窗，进行 sql 语句操作
? DataFrame 与 DataSet 支持一些特别方便的保存方式，比如保存成 csv，可以带上表
头，这样每一列的字段名一目了然(后面专门讲解) -
DataSet
? Dataset 和 DataFrame 拥有完全相同的成员函数，区别只是每一行的数据类型不同。
DataFrame 其实就是 DataSet 的一个特例 type DataFrame = Dataset[Row]
? DataFrame 也可以叫 Dataset[Row],每一行的类型是 Row，不解析，每一行究竟有哪
些字段，各个字段又是什么类型都无从得知，只能用上面提到的 getAS 方法或者共
性中的第七条提到的模式匹配拿出特定字段。而 Dataset 中，每一行是什么类型是
不一定的，在自定义了 case class 之后可以很自由的获得每一行的信息
三者的转换
GIt Commit
Commit 规范
使用Git Commit Guidelines作为规范方案
Commit Message 格式
<type>(<scope>): <subject>
<空行>
<body>
<空行>
<footer>
Type
feat：新功能（feature）
fix：修补bug
docs：文档（documentation）
style： 格式（不影响代码运行的变动）
refactor：重构（即不是新增功能，也不是修改bug的代码变动）
test：增加测试
chore：构建过程或辅助工具的变动
Scope
用来说明本次Commit影响的范围，即简要说明修改会涉及的部分.实际项目中可以看出基本上也成了必填项了。
Subject
用来简要描述本次改动，概述就好了，因为后面还会在Body里给出具体信息。并且最好遵循下面三条:
以动词开头，使用第一人称现在时，比如change，而不是changed或changes 首字母不要大写 结尾不用句号(.)
Body
里的内容是对上面subject里内容的展开，在此做更加详尽的描述，内容里应该包含修改动机和修改前后的对比。
Footer
footer里的主要放置不兼容变更和Issue关闭的信息