轉載自??2.數據湖DeltaLake之DDL操作

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

對創建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. 創建表

deltalake創建一張表有兩種方式：

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操作來創建表，CREATE TABLE.

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

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

?

2.數據分區

生產中使用構建數倉的時候會對數據進行分區操作，加速查詢，優化DML操作。使用Delta lake創建分區表的時候，只需要指定一個分區列即可。下面是一個按照常見的按照時間進行分區的例子：

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 metastore df.write.format("delta").partitionBy("date").save("/delta/events") // create table by path

3.指定存儲位置

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

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

假設創建Delta lake表的時候，指定的路徑里數據文件已經存在，創建的時候delta lake 會做以下事情：

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

CREATE TABLE eventsUSING DELTALOCATION '/delta/events'

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

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

3.讀數據

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

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會自動讀取最新的數據快照，用戶不需要進行refresh table。當可以使用謂詞下推的時候 delta lake會自動使用分區器和統計信息進行優化查詢，進而減少數據加載。

?

4.寫數據

a).Append

spark自身的append模式就可以完成往已有表里追加數據：???????

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

當然，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的時候，也可以支持只是覆蓋指定分區的數據。下面的例子就是只覆蓋一月份的數據：???????

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

這段代碼會自動將數據以dataframe的形式寫出到指定的deltalake表分區里，然后執行原子替換。

注意：

默認delta lake只支持數據的覆蓋，不支持schema的覆蓋替換。

總結

以上是生活随笔為你收集整理的2.数据湖DeltaLake之DDL操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。