155-029-33391

推广

首页 > 行业资讯 > 推广 > 正文内容
推广

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

iseeyu3年前 (2024-02-21)推广126

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>
  1. 更新操作
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>

扫描二维码推送至手机访问。

版权声明：本文由西安泽虎代运营发布，如需转载请注明出处。

转载请注明出处https://www.0291.com.cn/post/57061.html

返回列表

上一篇：大数据开发：Spark框架Dataset API

下一篇：HDFS入门介绍—(6千字总结)

相关文章

西安企业做百度推广，一个月花三千和花三万到底差在哪？

4|最近跟几个西安的老板聊天，发现一个挺有意思的现象：同样做百度推广，有人一个月花三千块，询盘不断；有人一个月砸三万，水花都没见着。问他们区别在哪，大多数人说不清楚，只知道'钱花了，没效果'。 5| 6|百度推广这东西，说复杂也复杂，说简单也简单。核心就一件事：让你的潜在客户搜某个词的时候，你的信息...

我想当策划，大学应该选什么专业?

我想当策划，大学应该选什么专业?

学，广告，传播，公关，这些专业其实都可以。另外也可以学像汉语言文学，以及一些跟文学相关的一些专业，会对你找工作有帮助。但是，你想从事这个这份职业，除了专业的学习之外，更多的是，你要在上学的时候积极参与一些更营销策划相关的社会实践。比如说你可以做一些兼职，比如说帮别人去写稿，...

服装市场营销策略分析，“定位不慎满盘皆输”

服装市场营销策略分析，“定位不慎满盘皆输”

不同的区域需要不同的策略，而不同的营销策略需要不同的销售思路来配合。因此，在研究女装产品销售思路之前，最好先对于市场与营销策略做一次深入地了解。1.市场区域化（market segmentation）：目标营销的第一步，是将整个市场拆分成若干个同性质的小市场。2.目标市场的...

晚间悦读丨你的思维方式，决定你的人生高度！

晚间悦读丨你的思维方式，决定你的人生高度！

越长大，越发现，能够决定你人生高度的，往往是你的认知层次。 你的认知在哪个层级，你的人生就处在什么状态。你认知的高度，决定了你人生的高度。 那如何才能升级方式、提升认知能力呢？可以从这三个方面做出改变。 NO. 1 改变态度 有人...

编写项目营销策划方案六大要点

编写项目营销策划方案六大要点

现在商业发展非常迅速，的需求非常的大，但是要做好营销策划是有一定难度的，我把自己做营销策划的一些体会在这里分享给朋友们。营销策划方案六大要点我们先说项目营销策划方案这个话题。大部分的营销策划人都是在企业里工作，大部分项目之初都是来源于一个想法，创始人、老板或者创始团队的一个...

【商场营销策划】如何进行商场营销策划？

【商场营销策划】如何进行商场营销策划？

商场是策划者们关注的永恒话题，那么究竟该如何作好商场营销策划呢？掌握以下这几点营销策略设计原则，至关重要！1.义利兼顾的原则。商场在获取利润的同时要充分考虑到消费者的合法利益和商场的长远利益。在作商场活动营销策略方案的时候应该将这一点作为首要的原则。对商场有利而对顾客不利的...

现在，非常期待与您的又一次邂逅

我们努力让每一部企业宣传片和抖音短视频成为商业大片

咨询热线 立即免费咨询
155-029-33391

Copyright 西安泽虎短视频代运营 Rights Reserved.陕ICP备2023011092号

Powered By Z-BlogPHP. Theme by TOYEAN.