Use ApsaraDB for SelectDB to query Hudi data source-ApsaraDB for SelectDB(SelectDB)-阿里云帮助中心

ApsaraDB for SelectDB supports federated queries against Apache Hudi data lakes through a catalog. This lets you run snapshot queries, time travel queries, and read optimized queries directly from SelectDB without moving data.

Use cases

Scenario	Description
Query acceleration	Use SelectDB's distributed engine to query Hudi data directly.

Supported query types

The queries available depend on the Hudi table type.

Table type	Supported queries
Copy On Write (COW)	Snapshot query, time travel query
Merge On Read (MOR)	Snapshot query, read optimized query, time travel query

Prerequisites

Before you begin, make sure that:

All nodes in your Hudi cluster are in the same virtual private cloud (VPC) as the SelectDB instance, or are connected to it. If they are in different VPCs, see What do I do if a connection fails to be established between an ApsaraDB for SelectDB instance and a data source?
The IP addresses of all Hudi cluster nodes are added to the IP address whitelist of the SelectDB instance. For more information, see Configure an IP address whitelist.
The IP addresses in the SelectDB VPC are added to the Hudi cluster's IP address whitelist, if the Hudi cluster supports whitelisting. To get the SelectDB instance's VPC IP addresses, see How do I view the IP addresses in the VPC to which my ApsaraDB SelectDB instance belongs?
You are familiar with catalog concepts and operations. For background, see Data lakehouse.

How it works

Hudi stores its metadata in Hive Metastore (HMS), which lets Hive identify and query Hudi tables. SelectDB integrates with Hudi through this same HMS layer — the integration process is identical to connecting a Hive data source. For details, see Hive data source.

Every write to a Hudi table creates a new snapshot. By default, SELECT * FROM <table_name> reads only the most recent snapshot.

Query Hudi data

Run a snapshot query

After you create a catalog pointing to your Hudi HMS, query a table as you would any SelectDB table:

SELECT * FROM <table_name>;

Run a time travel query

Use the FOR TIME AS OF clause to read historical data from a specific snapshot:

-- Timestamp format: "yyyy-MM-dd HH:mm:ss"
SELECT * FROM hudi_tbl FOR TIME AS OF "2022-10-07 17:20:37";

-- Compact timestamp format: "yyyyMMddHHmmss"
SELECT * FROM hudi_tbl FOR TIME AS OF "20221007172037";

Iceberg does not support the FOR TIME AS OF clause, unlike Hudi.

For supported timestamp formats, see Time Travel Query in the Apache Hudi documentation.

Column type mapping

The column data type mappings between SelectDB and Hudi are the same as those between SelectDB and Hive Catalog. For more information, see Column data type mappings.

Skip merge for MOR tables

When Apache Spark creates a Hudi Merge On Read (MOR) table, it also creates a read optimized table with the _ro suffix. SelectDB identifies read optimized tables using Hive InputFormat information — not the _ro suffix. Run SHOW CREATE TABLE to compare the InputFormat values across COW, MOR, and read optimized tables.

To skip the merge of log files when reading an MOR table, add hoodie.datasource.merge.type=skip_merge to the catalog properties. This setting is compatible with Spark Datasource Configs.

Query performance analysis

SelectDB uses different readers depending on the Hudi table type:

COW tables: read by the Parquet Native Reader (optimal performance)
MOR tables: read by the SDK for Java via Java Native Interface (JNI)

If you encounter a performance bottleneck, check whether the MOR read path is the cause.

Step 1: Check how many files use the Java SDK

Run EXPLAIN on your query to view the execution plan. The hudiNativeReadSplits parameter shows how many files the Parquet Native Reader handles versus the total file count. Files not handled by the Parquet Native Reader are read by the SDK for Java.

EXPLAIN SELECT * FROM <mor_table_name> WHERE <condition>;

Example output:

|0:VHUDI_SCAN_NODE                                                             |
|      table: minbatch_mor_rt                                                  |
|      predicates: `o_orderkey` = 100030752                                    |
|      inputSplitNum=810, totalFileSize=5645053056, scanRanges=810             |
|      partition=80/80                                                         |
|      numNodes=6                                                              |
|      hudiNativeReadSplits=717/810                                            |

In this example, 810 total files exist and 717 are read by the Parquet Native Reader. The remaining 93 files (810 − 717) are read by the SDK for Java.

Step 2: Profile the Java SDK read performance

If a large number of files are handled by the SDK for Java, run the following statement to check the query profile:

SHOW QUERY PROFILE "/"\G;

In the result, check these parameters. For more information, see Query profile.

Parameter	Description
`OpenScannerTime`	Time to create and initialize the JNI reader
`JavaScanTime`	Time the SDK for Java uses to read data
`FillBlockTime`	Time to copy Java data to C++ data
`GetRecordReaderTime`	Time the SDK for Java uses to create the Hudi record reader

What's next

Hive data source — understand the catalog integration model that Hudi builds on
Query profile — learn how to interpret full query profile output