数据湖—Delta Lake -之基础操作

1.数据湖的简单介绍：

1.1 官网

https://delta.io/

看一张官网的图

datalake.png

1.2 特点：

1.不限格式，来之不拒，均可流入
2.集中存储、到处可访问。
3.高性能分析能力 -- 借助于Spark、MR、SparkSQL等高性能分析计算引擎，可以对海量的数据进行分析。
4.原始数据存储
5.数据湖是一个存储企业的各种各样原始数据的大型仓库，其中的数据可供存取、处理、分析及传输。

1.3 数据湖，数据仓库， 数据集市 的对比

比较	数据仓库	数据集市	数据湖
应用范围	全公司	部门或者小组	全公司
数据类型	结构化数据处理	结构化数据处理	任意格式数据处理
存储规模	大量	中等规模(小型数仓)	海量
数据应用	维度建模，指标分析	小范围的数据分析	海量任意格式分析，不限应用类型
新应用开发周期	长	长	短

1.3 写时的模式

数据在写入之前，就需要定义好数据的schema，数据按照schema的定义写入

1.4 读时模式

数据在写入的时候，不需要定义Schema，在需要使用的时候在使用Schema定义它写时模式和读时模式是两种截然不同的数据处理方法。

数据湖就是一种读时模式思想的具体体现

1.相比较写时模式而言，读时模式因为是数据在使用到的时候再定义模型结构（Schema），因此能够提高数据模型定义的灵活性，可以满足不同上层业务的高效率分析需求。
2.因为，对于写时模式而言，如果想要事后更改Schema是有很高的成本的。
3.而读时模式可以在用的时候再定义Schema就很灵活了，同一套数据可以用不同的Schema来定义，来获取不同的效。

1.5 特点：

1.轻松的收集数据（读时模式）：数据湖与数据仓库的一大区别就是，Schema On Read，即在使用数据时才需要Schema信息；而数据仓库是Schema On Write，即在存储数据时就需要设计好Schema。这样，由于对数据写入没有限制，数据湖可以更容易的收集数据。
2.不需要关心数据结构：存储数据无限制，任意格式数据均可存储，只要你能分析就能存。
3.全部数据都是共享的（集中存储），多个业务单元或者研究人员可以使用全部的数据，以前由于一些数据分布于不同的系统上，聚合汇总数据是很麻烦的。
4.从数据中发掘更多价值（分析能力）：数据仓库和数据市场由于只使用数据中的部分属性，所以只能回答一些事先定义好的问题；而数据湖存储所有最原始、最细节的数据，所以可以回答更多的问题。并且数据湖允许组织中的各种角色通过自助分析工具（MR、Spark、SparkSQL等），对数据进行分析，以及利用AI、机器学习的技术，从数据中发掘更多的价值。
5.具有更好的扩展性和敏捷性：数据湖可以利用分布式文件系统来存储数据，因此具有很高的扩展能力。开源技术的使用还降低了存储成本。数据湖的结构没那么严格，因此天生具有更高的灵活性，从而提高了敏捷性。

1.6 数据湖的要求

1.安全：数据集中存储，就对数据安全有了更高的要求，对权限的管控要求更加严格。
2.可拓展的：随着业务扩张、数据增多，要求数据湖体系可以随需求扩展其能力。
3.可靠的：作为一个集中存储的数据中心，可靠性也很重要，三天两头坏掉那是不可以的。
4.吞吐量：数据湖作为海量数据的存储，对数据的吞吐量要求就必须很高。
5.原有格式存储：数据湖我们定义为 所有数据的原始数据集中存储库，那么存储进入数据湖的数据就是未经修饰的、原始的数据
6.支持多种数据源的输入：不限制数据类型，任意数据可以写入
7.多分析框架的支持：因为数据格式各种各样，并不全是结构化数据，所以，要求支持多种分析框架对数据湖中的数据进行提取、分析。包括但不限于：批处理的、实时的、流的、机器学习的、图形计算的等等。

1.7数据湖的原则

    1. 分离数据 和 业务
    2.存储和计算的分离（可选，比较适用云平台）
    3.Lambda架构 VS Kappa架构 VS IOTA架构 - 
    4.管理服务的重要性和选择合适的工具
        4.1安全 (Kerberos)
        4.2权限(Ranger)

2.Data Lake 的基本操作

2.1 Data Lake 的特点

1.  ACID 事务控制 ：Delta Lake将ACID事务带入您的数据湖。它提供了可序列化性，最强的隔离级别。
2.  可伸缩的元数据处理： Delta Lake可以轻松处理具有数十亿个分区和文件的PB级表
4.  数据版本控制 ： Delta Lake提供了数据快照，使开发人员可以访问和还原到较早版本的数据以进行审核，回滚或重现实验。
5.  开放的数据格式 ：Delta Lake中的所有数据均以Apache Parquet格式存储，从而使Delta Lake能够利用Parquet固有的高效压缩和编码方案。
6.  统一的批处理和流处理的source 和 sink ： Delta Lake中的表既是批处理表，又是流计算的source 和 sink。
7.  Schema执行： Delta Lake提供了指定和执行模式的功能。这有助于确保数据类型正确并且存在必需的列，从而防止不良数据导致数据损坏.
8.  Schema演化: 大数据在不断变化。 Delta Lake使您可以更改可自动应用的表模式，而无需繁琐的DDL
9.  审核历史记录 :Delta Lake事务日志记录有关数据所做的每项更改的详细信息，从而提供对更改的完整审核跟踪
10. 更新和删除 : Delta Lake支持Scala / Java API进行合并，更新和删除数据集。
10.100%和 Apache Spark 的API兼容    : 和spark 完全兼容。

2.2 Data lake 的操作： Spark Scala Shell -- 要求只是使用的Spark版本：>=2.4.2

bin/spark-shell --packages io.delta:delta-core_2.11:0.5.0

操作如图：

[root@master01 spark-2.4.7-bin-hadoop2.7]# bin/spark-shell --packages io.delta:delta-core_2.11:0.5.0
Ivy Default Cache set to: /root/.ivy2/cache
The jars for the packages stored in: /root/.ivy2/jars
:: loading settings :: url = jar:file:/opt/module/spark-2.4.7-bin-hadoop2.7/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
io.delta#delta-core_2.11 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-811cb329-3b4b-4a62-ab7a-d2287a1901dc;1.0
    confs: [default]
    found io.delta#delta-core_2.11;0.5.0 in central
    found org.antlr#antlr4;4.7 in central
    found org.antlr#antlr4-runtime;4.7 in central
    found org.antlr#antlr-runtime;3.5.2 in central
    found org.antlr#ST4;4.0.8 in central
    found org.abego.treelayout#org.abego.treelayout.core;1.0.3 in central
    found org.glassfish#javax.json;1.0.4 in central
    found com.ibm.icu#icu4j;58.2 in central
:: resolution report :: resolve 376ms :: artifacts dl 6ms
    :: modules in use:
    com.ibm.icu#icu4j;58.2 from central in [default]
    io.delta#delta-core_2.11;0.5.0 from central in [default]
    org.abego.treelayout#org.abego.treelayout.core;1.0.3 from central in [default]
    org.antlr#ST4;4.0.8 from central in [default]
    org.antlr#antlr-runtime;3.5.2 from central in [default]
    org.antlr#antlr4;4.7 from central in [default]
    org.antlr#antlr4-runtime;4.7 from central in [default]
    org.glassfish#javax.json;1.0.4 from central in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   8   |   0   |   0   |   0   ||   8   |   0   |
    ---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-811cb329-3b4b-4a62-ab7a-d2287a1901dc
    confs: [default]
    0 artifacts copied, 8 already retrieved (0kB/9ms)
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
21/06/09 23:27:26 WARN util.Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Spark context Web UI available at http://master01.pxx.com:4041
Spark context available as 'sc' (master = local[*], app id = local-1623252446880).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.7
      /_/
         
Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_251)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val data = spark.range(0, 5)

2.2 官网命令：

bin/spark-shell --packages io.delta:delta-core_2.12:1.0.0 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"
其实可以自行bin/spark-shell --packages io.delta:delta-core_2.12:1.0.0

2.3 按照官网命令走

1.创建表， 并且读取表

scala> val data = spark.range(0, 5)
data: org.apache.spark.sql.Dataset[Long] = [id: bigint]
scala> data.write.format("delta").save("/tmp/delta-table02")                                                                         
scala> spark.read.format("delta").load("/tmp/delta-table02").toDF.show()
+---+
| id|
+---+
|  2|
|  0|
|  4|
|  3|
|  1|
+---+
scala> 

2. 更新操作

scala> val data01 = spark.range(5,10)
data01: org.apache.spark.sql.Dataset[Long] = [id: bigint]
scala> data01.write.format("delta").mode("overwrite").save("/tmp/delta-table02")                                                                             
scala> spark.read.format("delta").load("/tmp/delta-table02").toDF.show()
+---+
| id|
+---+
|  8|
|  7|
|  5|
|  6|
|  9|
+---+
scala> 

3.Delta Lake提供了编程api，用于有条件地更新、删除和合并(upsert)数据到表中

scala> import io.delta.tables._
import io.delta.tables._

scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._

scala> val deltaTable = DeltaTable.forPath("/tmp/delta-table02")
deltaTable: io.delta.tables.DeltaTable = io.delta.tables.DeltaTable@5e88e2e7

// 通过将每个偶数值加100来更新每个偶数值
scala> deltaTable.update(condition=expr("id % 2 ==0"), set = Map("id"->expr("id+100")))
                                                                                
scala> spark.read.format("delta").load("/tmp/delta-table02").toDF.show()
+---+                                                                           
| id|
+---+
|106|
|  7|
|  5|
|108|
|  9|
+---+

// 删除偶数
scala> deltaTable.delete(condition = expr("id % 2 ==0"))
                                                                                
scala> spark.read.format("delta").load("/tmp/delta-table02").toDF.show()
+---+                                                                           
| id|
+---+
|  7|
|  5|
|  9|
+---+


scala> val newData = spark.range(0,20).toDF
newData: org.apache.spark.sql.DataFrame = [id: bigint]
// 合并新数据
 deltaTable.as("oldData").merge(newData.as("newData"),"oldData.id=newData.id").whenMatched.update(Map("id" -> col("newData.id"))).whenNotMatched.insert(Map("id" ->col("newData.id"))).excute()
                                                                                                                                                                                             ^

scala> deltaTable.as("oldData").merge(newData.as("newData"),"oldData.id=newData.id").whenMatched.update(Map("id" -> col("newData.id"))).whenNotMatched.insert(Map("id" ->col("newData.id"))).execute()
[Stage 86:===================================>                 (135 + 51) / 200]21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 96.54% for 7 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 84.47% for 8 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 75.08% for 9 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 67.58% for 10 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 61.43% for 11 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 56.31% for 12 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 51.98% for 13 writers
[Stage 86:===============================================>     (180 + 20) / 200]21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 48.27% for 14 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 45.05% for 15 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 42.24% for 16 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 39.75% for 17 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 37.54% for 18 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 35.57% for 19 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 33.79% for 20 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 35.57% for 19 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 37.54% for 18 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 39.75% for 17 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 42.24% for 16 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 45.05% for 15 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 48.27% for 14 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 51.98% for 13 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 56.31% for 12 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 61.43% for 11 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 67.58% for 10 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 75.08% for 9 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 84.47% for 8 writers
21/06/09 23:51:11 WARN hadoop.MemoryManager: Total allocation exceeds 95.00% (906,992,014 bytes) of heap memory
Scaling row group sizes to 96.54% for 7 writers
                                                                                
scala> deltaTable.toDF.show()
+---+                                                                           
| id|
+---+
|  0|
|  2|
|  6|
|  1|
| 10|
| 11|
| 15|
| 12|
|  4|
| 19|
| 14|
|  5|
|  9|
| 13|
|  8|
| 18|
| 16|
|  7|
|  3|
| 17|
+---+
scala>