E-MapReduce:Tiered Storage of Hot and Cold Data

Prerequisites

Before you begin, ensure that you have:

• An EMR cluster with Streaming SQL enabled

• A Tablestore instance and the permissions to create tables and tunnels

• The Tablestore SDK for Java (for writing sample data)

Step 1: Set up the source table and load sample data

Create a Tablestore table named OrderSource with two primary key columns (UserId and OrderId) and two attribute columns (price and timestamp).

Use the BatchWriteRow operation from the Tablestore SDK for Java to write order data into the table. In this example, the timestamp range covers the latest 90 days (2020-02-26 to 2020-05-26), and the total number of records is 3,112,400.

When writing data, populate the timestamp column as the version number attribute column of the Tablestore row. Configure TTL on the table so that rows whose version number exceeds the retention period are automatically deleted.

Step 2: Create a CDC tunnel

In the Tablestore console, create a tunnel on the OrderSource table for incremental synchronization. Record the tunnel ID — you will use it in the Streaming SQL configuration.

Step 3: Start Streaming SQL and create source and sink tables

On the Header node of your EMR cluster, start the Streaming SQL interactive shell:

streaming-sql --master yarn --use-emr-datasource --num-executors 16 --executor-memory 4g --executor-cores 4

Run the following commands to create the Tablestore source table, the Delta Lake sink, the incremental scan view, and the streaming job:

-- 1. Create the Tablestore source table.
DROP TABLE IF EXISTS order_source;
CREATE TABLE order_source
USING tablestore
OPTIONS(
  endpoint="http://vehicle-test.cn-hangzhou.vpc.tablestore.aliyuncs.com",
  access.key.id="",
  access.key.secret="",
  instance.name="vehicle-test",
  table.name="OrderSource",
  catalog='{"columns": {"UserId": {"col": "UserId", "type": "string"}, "OrderId": {"col": "OrderId", "type": "string"}, "price": {"col": "price", "type": "double"}, "timestamp": {"col": "timestamp", "type": "long"}}}'
);

-- 2. Create the Delta Lake sink.
DROP TABLE IF EXISTS delta_orders;
CREATE TABLE delta_orders(
  UserId string,
  OrderId string,
  price double,
  timestamp long
)
USING delta
LOCATION '/delta/orders';

-- 3. Create an incremental scan view for the source table.
--    tunnel.id: the ID of the tunnel created in Step 2.
--    maxoffsetsperchannel: the maximum data volume that can be written to each partition over the tunnel.
CREATE SCAN incremental_orders ON order_source USING STREAM
OPTIONS(
  tunnel.id="324c6bee-b10d-4265-9858-b829a1b71b4b",
  maxoffsetsperchannel="10000"
);

-- 4. Start the streaming job.
--    The job merges CDC changes (PUT / UPDATE / DELETE) into delta_orders.
--    __ots_record_type__ is a predefined column from the Tablestore streaming source
--    that identifies the row operation type.
CREATE STREAM orders_job
OPTIONS (
  checkpointLocation='/delta/orders_checkpoint',
  triggerIntervalMs='3000'
)
MERGE INTO delta_orders
USING incremental_orders AS delta_source
ON delta_orders.UserId=delta_source.UserId AND delta_orders.OrderId=delta_source.OrderId
WHEN MATCHED AND delta_source.__ots_record_type__='DELETE' THEN
  DELETE
WHEN MATCHED AND delta_source.__ots_record_type__='UPDATE' THEN
  UPDATE SET UserId=delta_source.UserId, OrderId=delta_source.OrderId, price=delta_source.price, timestamp=delta_source.timestamp
WHEN NOT MATCHED AND delta_source.__ots_record_type__='PUT' THEN
  INSERT (UserId, OrderId, price, timestamp) VALUES (delta_source.UserId, delta_source.OrderId, delta_source.price, delta_source.timestamp);

The four SQL statements do the following:

Step 4: Verify initial sync and set the TTL

Before changing the TTL, confirm that both tables contain the same number of records.

Query record counts in both order_source and delta_orders. The counts should match (3,112,400 records each).

Set the TTL for the OrderSource table to 30 days. After the TTL takes effect:

• order_source retains only records from the latest 30 days: 1,017,004 records.

• delta_orders still contains all records: 3,112,400 records.

Hot data and cold data are now separated.