Hudi is a data lake framework that allows you to update and delete data in Hadoop-compatible file systems and consume changed data. E-MapReduce (EMR) Presto integrates the related JAR packages into a standalone Hudi plugin. You can use the EMR Hudi connector to query Copy on Write (COW) and Merge on Read (MOR) tables.
Background information
For more information about EMR Hudi, see Hudi overview.
Prerequisites
A Hadoop cluster that includes the Presto service or a standalone Presto cluster is created. For more information, see Create a cluster.
Limits
- The Hudi connector is supported only by Hadoop or Presto clusters of EMR V3.38.0 or later.
- For COW tables, only snapshot queries are supported.
- For MOR tables, only snapshot queries and read-optimized queries are supported.
- Incremental queries are not supported.
Configure the Hudi connector
Modify the Hudi connector configuration. For more information, see Modify a built-in connector.
To view the default Hudi connector configuration, go to the Presto service page in the EMR console. On the Configure tab, find the Service Configuration section and click the hudi.properties tab. The following table describes the parameters. Modify the parameter values based on your business requirements.
| Parameter | Description |
| hive.recursive-directories | Enables reading data from subdirectories of table or partition locations. This is similar to the hive.mapred.supports.subdirectories property in Hive. |
| hive.metastore.uri | The URI for connecting to the Hive Metastore using the Thrift protocol. The default value is in the format |
| hive.config.resources | A comma-separated list of HDFS configuration files. These files must exist on all Presto hosts. Important Configure this parameter only when access to HDFS is required. |
| hive.delta-table-enabled | Specifies whether Presto can read Delta Lake tables. The valid values are:
|
| hive.delta-compatible-mode-enabled | Specifies whether to enable compatibility mode for Delta Lake tables. The valid values are:
|
| hive.hdfs.impersonation.enabled | Specifies whether to enable user impersonation. The valid values are:
|
Example
Hudi tables can be stored as external Hive tables. You can use the Hive connector to query data in Hudi tables. For information about how to create Hudi tables and synchronize them to Hive tables, see Integrate Hudi with Spark SQL and Write data to Hudi by using Spark.
The following example shows how to generate and query data:
- Log on to the EMR cluster in SSH mode. For more information, see Log on to a cluster.
- Run the following command to open the Spark SQL CLI:
spark-sql --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'If the output contains the following information, the Spark SQL CLI is opened:spark-sql> - Run the following command to create a test table named emr_test:
create table if not exists emr_test( id bigint, name string, price double ) using hudi options ( type = 'mor', primaryKey = 'id,name' ); - Run the following commands to insert test data:
insert into emr_test select 1, 'a2', 10; insert into emr_test select 1, 'a1', 10; insert into emr_test select 2, 'a1', 20;Note EMR Spark SQL automatically synchronizes Hudi data to Data Lake Formation (DLF) or the Hive Metastore. - Query data in the Presto client.
- Run the following command to open the Presto client:
presto --server emr-header-1:9090 --catalog hudi --schema default --user hadoop - Run the following command to query the table:
select * from emr_test;The following output is returned:_hoodie_commit_time | _hoodie_commit_seqno | _hoodie_record_key | _hoodie_partition_path | _hoodie_file_name | id | name | price ---------------------+----------------------+--------------------+------------------------+-------------------------------------------------------------------------+----+------+------- 20211025145616 | 20211025145616_0_1 | id:1,name:a2 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-20-1604_20211025145616.parquet | 1 | a2 | 10.0 20211025145629 | 20211025145629_0_1 | id:1,name:a1 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-48-3211_20211025145629.parquet | 1 | a1 | 10.0 20211025145640 | 20211025145640_0_2 | id:2,name:a1 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-76-4818_20211025145640.parquet | 2 | a1 | 20.0 (3 rows)
- Run the following command to open the Presto client:
- Update data in the Spark SQL CLI.
- Run the following command to open the Spark SQL CLI:
spark-sql --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \ --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'If the output contains the following information, the Spark SQL CLI is opened:spark-sql> - Run the following command to update price for the row whose id is 2:
update emr_test set price = price + 20 where id = 2;
- Run the following command to open the Spark SQL CLI:
- After the update, query the data in the Presto client.
- Run the following command to open the Presto client:
presto --server emr-header-1:9090 --catalog hudi --schema default --user hadoop - Run the following command to query the table:
select * from emr_test;The following output is returned:_hoodie_commit_time | _hoodie_commit_seqno | _hoodie_record_key | _hoodie_partition_path | _hoodie_file_name | id | name | price ---------------------+----------------------+--------------------+------------------------+-------------------------------------------------------------------------+----+------+------- 20211025145616 | 20211025145616_0_1 | id:1,name:a2 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-20-1604_20211025145616.parquet | 1 | a2 | 10.0 20211025145629 | 20211025145629_0_1 | id:1,name:a1 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-48-3211_20211025145629.parquet | 1 | a1 | 10.0 20211025145640 | 20211025145640_0_2 | id:2,name:a1 | | ac4ec1e6-528d-4189-bde6-d09e137f63f6-0_0-76-4818_20211025145640.parquet | 2 | a1 | 40.0 (3 rows)
- Run the following command to open the Presto client: