數(shù)據(jù)湖DeltaLake中的DDL操作是怎么實現(xiàn)的

本篇文章為大家展示了數(shù)據(jù)湖DeltaLake中的DDL操作是怎么實現(xiàn)的，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

前面講了delta lake簡介，特性及基本操作。本文主要是講DeltaLake的DDL操作，實際上是依賴于spark datasourcev2 和catalog API（3.0+）的，所以Deltalake整合spark的時候最好是3.0開始吧，正好最近spark 3.0也發(fā)布了。

對創(chuàng)建sparksession有點要求，需要加上兩個配置：

valspark = SparkSession  .builder()  .appName(this.getClass.getCanonicalName)  .master("local[2]")  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") .getOrCreate()

1. 創(chuàng)建表

deltalake創(chuàng)建一張表有兩種方式：

1). DataFrameWriter，這個大家不陌生，spark默認寫文件的方式。

df.write.format("delta").saveAsTable("events")      // create table in the metastore
df.write.format("delta").save("/delta/events")  // create table by path

2). DeltaLake也支持使用spark sql新的DDL操作來創(chuàng)建表，CREATE TABLE.

-- Create table in the metastoreCREATE TABLE events (  date DATE,  eventId STRING,  eventType STRING,  data STRING)USING DELTA

?

當(dāng)創(chuàng)建使用Deltalake在metastore中創(chuàng)建一張表的時候，會在metastore中記錄數(shù)據(jù)的位置信息。這樣好處很明顯，其他人使用的時候就比較方便查找,不用關(guān)注數(shù)據(jù)的真是存儲位置。然而，metastore里不會存儲數(shù)據(jù)內(nèi)容是否有效。

2.數(shù)據(jù)分區(qū)

生產(chǎn)中使用構(gòu)建數(shù)倉的時候會對數(shù)據(jù)進行分區(qū)操作，加速查詢，優(yōu)化DML操作。使用Delta lake創(chuàng)建分區(qū)表的時候，只需要指定一個分區(qū)列即可。下面是一個按照常見的按照時間進行分區(qū)的例子：

1).DDL操作

-- Create table in the metastoreCREATE TABLE events ( date DATE, eventId STRING, eventType STRING, data STRING)USING DELTAPARTITIONED BY (date)LOCATION '/delta/events'

2).Scala API

df.write.format("delta").partitionBy("date").saveAsTable("events")      // create table in the metastoredf.write.format("delta").partitionBy("date").save("/delta/events")  // create table by path

3.指定存儲位置

其實，我們可以控制Delta lake的表數(shù)據(jù)文件的存儲位置，在寫DDL的時候可以指定path。

這個其實很像hive 的外部表的功能，指定位置的delta lake的表，可以視為是不會被metastore管理的，這種表在刪除的時候，不會被真實刪除數(shù)據(jù)。

假設(shè)創(chuàng)建Delta lake表的時候，指定的路徑里數(shù)據(jù)文件已經(jīng)存在，創(chuàng)建的時候delta lake 會做以下事情：

1).如果你僅僅在創(chuàng)建的時候指定了表名稱和路徑，如下：

CREATE TABLE eventsUSING DELTALOCATION '/delta/events'

hive metastore的表會自動從存在的數(shù)據(jù)中推斷出schema，partition，和屬性。該功能可用于將數(shù)據(jù)導(dǎo)入metastore中。

2).假設(shè)你指定了一些配置（schema，partition，或者表的屬性），delta lake只會從已有數(shù)據(jù)中識別出你指定的配置信息，而不是全部配置。假設(shè)你指定的配置在已有數(shù)據(jù)中不存在，那么會拋出不一致異常。

3.讀數(shù)據(jù)

數(shù)據(jù)可以直接支持sql查詢，老spark用戶也可以直接使用dataframe api去查詢數(shù)據(jù)。

sql查詢

SELECT * FROM events   -- query table in the metastore
SELECT * FROM delta.`/delta/events`  -- query table by path

dataframe查詢

spark.table("events")      // query table in the metastore
spark.read.format("delta").load("/delta/events")  // create table by path

dataframe會自動讀取最新的數(shù)據(jù)快照，用戶不需要進行refresh table。當(dāng)可以使用謂詞下推的時候 delta lake會自動使用分區(qū)器和統(tǒng)計信息進行優(yōu)化查詢，進而減少數(shù)據(jù)加載。

4.寫數(shù)據(jù)

a).Append

spark自身的append模式就可以完成往已有表里追加數(shù)據(jù)：

df.write.format("delta").mode("append").save("/delta/events")df.write.format("delta").mode("append").saveAsTable("events")

當(dāng)然，delta 也支持insert into：

INSERT INTO events SELECT * FROM newEvents

b).overwrite

delta lake也支持直接覆蓋整張表，直接使用overwrite模式即可。

dataframe api如下：

df.write.format("delta").mode("overwrite").save("/delta/events")df.write.format("delta").mode("overwrite").saveAsTable("events")

SQL API格式如下：

INSERT OVERWRITE events SELECT * FROM newEvents

使用Dataframe的時候，也可以支持只是覆蓋指定分區(qū)的數(shù)據(jù)。下面的例子就是只覆蓋一月份的數(shù)據(jù)：

df.write  .format("delta")  .mode("overwrite")  .option("replaceWhere", "date >= '2017-01-01' AND date <= '2017-01-31'")  .save("/delta/events")

這段代碼會自動將數(shù)據(jù)以dataframe的形式寫出到指定的deltalake表分區(qū)里，然后執(zhí)行原子替換。

注意：

默認delta lake只支持數(shù)據(jù)的覆蓋，不支持schema的覆蓋替換。

上述內(nèi)容就是數(shù)據(jù)湖DeltaLake中的DDL操作是怎么實現(xiàn)的，你們學(xué)到知識或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識儲備，歡迎關(guān)注億速云行業(yè)資訊頻道。