E-MapReduce:Use Delta Lake

Step 1: Create an SQL session

1. Go to the Sessions page.

   1. Log on to the EMR console.

   2. In the left-side navigation pane, choose EMR Serverless > Spark.

   3. On the Spark page, click the name of the target workspace.

   4. On the EMR Serverless Spark page, click Sessions in the left-side navigation pane.

2. On the SQL Session page, click Connect to SQL Session.

3. On the Connect to SQL Session page, in the Spark Configuration section, add the following properties and click create. For more information, see Manage SQL sessions.

   The metadata is stored in the default catalog of the current workspace. If you want to change the default catalog to an external Hive Metastore, see Connect to an external Hive Metastore service.

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

Step 2: Read and write Delta Lake tables

1. Go to the SQL development page.

   On the EMR Serverless Spark page, click Development in the left-side navigation pane.

2. On the Development tab, click the icon.

3. In the Create dialog box, enter a name such as users_task, set the Type to the default SparkSQL, and then click OK.

4. Copy the following code to the new SparkSQL tab (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. From the drop-down lists, select a database and the SQL session that you just created.

6. Click Run.

   The query returns a table that contains two columns, id and name, and two records: id=1, name='a' and id=2, name='b'.

Step 3: Update data

Delta Lake supports various data manipulation language (DML) operations, including UPDATE, DELETE, and MERGE INTO. The following code provides examples.

-- update operation
UPDATE ss_delta_db.delta_tbl SET name = "a_v2" WHERE id = 1;
-- delete operation
DELETE FROM ss_delta_db.delta_tbl WHERE id = 2;
-- merge into operation
-- Create a temporary table and insert data into the table.
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");
-- Run the MERGE INTO operation.
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 *;
-- Verify the result.
SELECT * FROM ss_delta_db.delta_tbl ORDER BY id;

The query returns a table that contains two columns, id and name, and two records: id=1, name='a_v3' and id=3, name='c'.

Step 4: Clean up resources

After testing, clean up the resources to avoid storage fees. Run the following commands:

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