Manage Hudi data using DML statements in Spark SQL - E-MapReduce

This topic covers the Data Manipulation Language (DML) statements supported when Apache Hudi is integrated with Spark SQL on E-MapReduce (EMR): MERGE INTO, INSERT INTO, UPDATE, and DELETE.

Prerequisites

Before you begin, ensure that you have:

An EMR cluster with the Spark and Hudi services installed. For details, see Create a cluster

Limitations

Only the following cluster versions support using Spark SQL to read from and write to Hudi:

EMR V3.36.0 or later minor versions
EMR V5.2.0 or later minor versions

Start Spark SQL

The required startup flags depend on your Spark and Hudi versions.

Spark 2 or Spark 3 with Hudi earlier than 0.11

spark-sql \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

Spark 3 with Hudi 0.11 or later

spark-sql \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

MERGE INTO

Inserts, updates, or deletes rows in a target Hudi table based on a match condition against a source dataset.

Syntax

MERGE INTO tableIdentifier AS target_alias
USING (subquery | tableIdentifier) AS source_alias
ON <merge_condition>
WHEN MATCHED [ AND <condition> ] THEN <matched_action>
[ WHEN MATCHED [ AND <condition> ] THEN <matched_action> ]
[ WHEN NOT MATCHED [ AND <condition> ] THEN <not_matched_action> ]

<merge_condition>    = An equality condition
<matched_action>     =
  DELETE |
  UPDATE SET * |
  UPDATE SET column1 = value1 [, column2 = value2 ...]
<not_matched_action> =
  INSERT * |
  INSERT (column1 [, column2 ...]) VALUES (value1 [, value2 ...])

Examples

Upsert without delete

-- Update matching rows; insert non-matching rows
MERGE INTO h0 AS target
USING (
  SELECT 1 AS id, 'a1' AS name, 10.0 AS price
) source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Upsert with conditional delete

-- Update matching rows; delete rows where name = 'delete'; insert non-matching rows
MERGE INTO h0 AS target
USING (
  SELECT 1 AS id, 'a1' AS name, 10.0 AS price
) source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET id = source.id, name = source.name, price = source.price
WHEN MATCHED AND name = 'delete' THEN DELETE
WHEN NOT MATCHED THEN INSERT (id, name, price) VALUES (id, name, price);

INSERT INTO

Inserts rows into a partitioned or non-partitioned Hudi table.

Examples

Insert into a non-partitioned table

-- Insert into a non-partitioned Hudi table
INSERT INTO h0 SELECT 1, 'a1', 20;

Insert into a table with a static partition

-- Insert into a partitioned Hudi table using a static partition value
INSERT INTO h_p0 PARTITION(dt='2021-01-02') SELECT 1, 'a1';

Insert into a table with dynamic partitions (form 1)

-- Insert into a partitioned Hudi table; partition value comes from the source query
INSERT INTO h_p0 PARTITION(dt) SELECT 1, 'a1', dt FROM s;

Insert into a table with dynamic partitions (form 2)

-- Partition fields placed at the end of the SELECT list
INSERT INTO h_p1 SELECT 1 AS id, 'a1', '2021-01-03' AS dt, '19' AS hh;

Overwrite a table

-- Replace all existing data in h0
INSERT OVERWRITE TABLE h0 SELECT 1, 'a1', 20;

UPDATE

Updates one or more columns in rows that match a condition, in a partitioned or non-partitioned Hudi table.

Syntax

UPDATE tableIdentifier SET column = EXPRESSION [, column = EXPRESSION ...] [WHERE condition];

Example

-- Set price to 20 for the row where id = 1
UPDATE h0 SET price = 20 WHERE id = 1;

DELETE

Deletes rows that match a condition from a partitioned or non-partitioned Hudi table.

Syntax

DELETE FROM tableIdentifier [WHERE BOOL_EXPRESSION];

Example

-- Delete all rows where id > 100
DELETE FROM h0 WHERE id > 100;