使用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），类型使用默认的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;