E-MapReduce：使用Delta Lake

步驟一：建立SQL會話

1. 進入會話管理頁面。

   1. 登入E-MapReduce控制台。

   2. 在左側導覽列，選擇EMR Serverless > Spark。

   3. 在Spark頁面，單擊目標工作空間名稱。

   4. 在EMR Serverless Spark頁面，單擊左側導覽列中的會話管理。

2. 在SQL會話頁面，單擊建立SQL會話。

3. 在建立SQL會話頁面的Spark配置地區，配置以下資訊，單擊建立。詳情請參見管理SQL會話。

   中繼資料是當前工作空間的預設Catalog。如果您希望將預設Catalog修改為外部的Hive Metastore，可以參見串連外部Hive Metastore Service。

   spark.sql.extensions             io.delta.sql.DeltaSparkSessionExtension
   spark.sql.catalog.spark_catalog  org.apache.spark.sql.delta.catalog.DeltaCatalog

步驟二：讀寫Delta Lake表

1. 進入SQL開發頁面。

   在EMR Serverless Spark頁面，單擊左側導覽列中的開發。

2. 在開發目錄頁簽下，單擊表徵圖。

3. 在建立對話方塊中，輸入名稱（例如users_task），類型使用預設的SQL > SparkSQL，然後單擊確定。

4. 拷貝如下代碼到新增的SparkSQL頁簽（users_task）中。

   CREATE DATABASE IF NOT EXISTS ss_delta_db;
   
   CREATE TABLE ss_delta_db.delta_tbl (id INT, name STRING) USING delta;
   
   INSERT INTO ss_delta_db.delta_tbl VALUES (1, "a"), (2, "b");
   
   SELECT id, name FROM ss_delta_db.delta_tbl ORDER BY id;

5. 在資料庫下拉式清單中選擇一個資料庫，在會話下拉式清單中選擇剛剛建立的SQL會話。

6. 單擊運行，執行任務。返回資訊如下所示。

   

步驟三：更新操作

Delta Lake支援多種資料更新操作，包括Update、Delete和Merge Into。以下是具體樣本。

-- Update操作
UPDATE ss_delta_db.delta_tbl SET name = "a_v2" WHERE id = 1;

-- Delete操作
DELETE FROM ss_delta_db.delta_tbl WHERE id = 2;


-- Merge Into操作
-- 建立暫存資料表並插入資料
CREATE TABLE ss_delta_db.tmp_tbl(id INT, name STRING) USING delta;
INSERT INTO ss_delta_db.tmp_tbl VALUES (1, "a_v3"), (3, "c");

-- 執行Merge Into操作
MERGE INTO ss_delta_db.delta_tbl AS target
USING ss_delta_db.tmp_tbl AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

--驗證結果
SELECT * FROM ss_delta_db.delta_tbl ORDER BY id;

返回資訊如下所示。

步驟四：清理操作

測試完成後，建議清理測試資源以避免佔用儲存空間。執行以下命令如下所示。

DROP TABLE ss_delta_db.delta_tbl;
DROP TABLE ss_delta_db.tmp_tbl;

DROP DATABASE ss_delta_db;