E-MapReduce:Delta connector

Prerequisites

You have created one of the following cluster types: a DataLake or Custom cluster with the Trino service, or a Hadoop cluster with the Presto service. For more information, see Create a cluster.

Limitations

The Delta connector is supported on DataLake clusters, Custom clusters, and Hadoop clusters that run EMR-3.39.1 or later, or EMR-5.5.0 or later.

Basic usage

Modify connector configuration

To modify the Delta connector configuration, see Modify the configurations of a built-in connector.

Default connector configuration

Example

Trino cannot create or modify a Delta Lake table. Use Spark SQL for these operations instead. For more information, see Basic usage.

1. Generate data.

   1. Run the following command to enter the Spark SQL command line.

      spark-sql

   2. Run the following statement to create a Delta Lake table.

      CREATE TABLE delta_table (id INT) USING delta;

   3. Run the following statement to write data to the table.

      INSERT INTO delta_table VALUES 0,1,2,3,4;

2. Query the data.

   1. Access the Trino command line. For more information, see Access Trino using a command-line interface.

   2. Run the following statement to query the table.

      SELECT * FROM delta_table;

      The following output is returned.

       id
      ----
        0
        1
        2
        3
        4
      (5 rows)

Advanced usage

Time travel

The time travel feature lets you query a table's historical data.

Example:

1. Run the following command to enter the Spark SQL command line.

   spark-sql

2. Run the following statement to overwrite the data.

   INSERT OVERWRITE TABLE delta_table VALUES 5,6,7,8,9;

3. Query the data.

   1. Access the Trino command line. For more information, see Access Trino using a command-line interface.

   2. Run the following statement to query the table.

      SELECT * FROM delta_table;

      The following output is returned.

       id
      ----
        5
        6
        7
        8
        9
      (5 rows)

4. Use time travel to query historical data.

   Run the following statement to query data by version number. The version number is a monotonically increasing integer. By default, the version number is 1 after the first INSERT operation and increments by 1 with each modification.

   SELECT * FROM delta_table FOR VERSION AS OF 1;

   The following output is returned.

    id
   ----
     2
     1
     3
     4
     0
   (5 rows)

   You can also query data by timestamp. Three timestamp types are supported: DATE, TIMESTAMP, and TIMESTAMP WITH TIME ZONE.

   • DATE: Queries data at 00:00:00 UTC on the specified date.

   • TIMESTAMP: Queries data at the specified timestamp in UTC.

     For example, run the following statement to query data at 20:00:00 on February 15, 2022, in Beijing time (UTC+08:00), using the TIMESTAMP type.

     SELECT * FROM delta_table FOR TIMESTAMP AS OF TIMESTAMP '2022-02-15 12:00:00';

     Note

     In this statement, the first TIMESTAMP keyword sets the time travel mode to timestamp (not the version number mode). The second TIMESTAMP keyword casts the string literal to the TIMESTAMP data type (not the DATE type).

     The following output is returned.

      id
     ----
       2
       0
       3
       4
       1
     (5 rows)

   • TIMESTAMP WITH TIME ZONE: Querying with TIMESTAMP WITH TIME ZONE requires a type cast.

     For example, to query data at 20:00:00 on February 15, 2022, in Beijing time (UTC+08:00), use the following statement.

     SELECT * FROM delta_table FOR TIMESTAMP AS OF CAST('2022-02-15 20:00:00 +0800' AS TIMESTAMP WITH TIME ZONE);

Z-order

Trino uses Z-Ordering to optimize queries on Delta Lake tables by leveraging Parquet's native optimizations and data skipping. After optimization, Delta Lake collects file-level statistics such as minimum and maximum values for each column, which the Delta connector uses to filter data files directly.

For Delta Lake tables optimized with the OPTIMIZE and ZORDER BY commands, properly configured Z-Ordering columns can make Trino queries dozens of times faster.

Trino supports Z-Ordering for the following data types: Int, Long, Double, Float, Binary, Boolean, String, and Array.

For example, consider a table named conn_zorder with four columns: src_ip, src_port, dst_ip, and dst_port.

OPTIMIZE conn_zorder ZORDER BY (src_ip, src_port, dst_ip, dst_port);