Lindorm:Quick Start: Real-time ETL

Connect to LindormTable from the MySQL command line.

Create the source table and the mirror table.

-- Source table
CREATE TABLE source(p1 INT, c1 DOUBLE, PRIMARY KEY(p1));

-- Mirror (sink) table — same schema as the source
CREATE TABLE sink(p1 INT, c1 DOUBLE, PRIMARY KEY(p1));

Connect to the stream engine using the MySQL command line.

Create the ETL job to replicate all rows from source into sink.

CREATE ETL sync_etl AS INSERT INTO sink SELECT * FROM source;

Connect to LindormTable from the MySQL command line.

Insert rows into the source table.

INSERT INTO source(p1, c1) VALUES(0, 0.0);
INSERT INTO source(p1, c1) VALUES(1, 1.0);
INSERT INTO source(p1, c1) VALUES(2, 2.0);
INSERT INTO source(p1, c1) VALUES(3, 3.0);
INSERT INTO source(p1, c1) VALUES(4, 4.0);

Query the mirror table to confirm the rows were synced.

SELECT * FROM sink;

Expected output:

+------+------+
| p1   | c1   |
+------+------+
|    0 |  0.0 |
|    1 |  1.0 |
|    2 |  2.0 |
|    3 |  3.0 |
|    4 |  4.0 |
+------+------+

Connect to LindormTable from the MySQL command line.

Create the two source tables and a secondary index on order_tbl.

-- Source table 1: user information
CREATE TABLE `user_tbl` (
    `user_id`   varchar NOT NULL,
    `user_name` varchar,
    `user_addr` varchar,
    PRIMARY KEY (`user_id`)
);

-- Source table 2: order information
CREATE TABLE `order_tbl` (
    `order_id`     varchar NOT NULL,
    `user_id`      varchar,
    `product_name` varchar,
    `price`        decimal(38, 20),
    PRIMARY KEY (`order_id`)
);

-- Secondary index on order_tbl to support the JOIN on user_id.
-- Without this index, the ETL engine must perform a full table scan
-- on order_tbl for every incoming user_tbl write.
CREATE INDEX idx1 ON `order_tbl`(user_id desc) WITH (COMPRESSION='ZSTD');

Create the sink table to store the pre-joined data.

-- Sink table: combines user and order data, plus an extracted postal code.
-- MUTABILITY='MUTABLE_UDT' allows partial-column updates when only one
-- source table changes (for example, a user updates their address).
CREATE TABLE `user_order_tbl` (
    `order_id`       varchar NOT NULL,
    `user_id`        varchar,
    `product_name`   varchar,
    `price`          decimal(38, 20),
    `user_name`      varchar,
    `user_addr`      varchar,
    `user_addr_code` varchar,
    PRIMARY KEY (`order_id`)
) WITH (MUTABILITY = 'MUTABLE_UDT');

Connect to the stream engine using the MySQL command line.

Create the ETL job with the join and transformation logic.

CREATE ETL join_etl AS
INSERT INTO `lindorm_table`.`default`.`user_order_tbl` (
    order_id,
    user_id,
    product_name,
    price,
    user_name,
    user_addr,
    user_addr_code
)
SELECT
    o.order_id,
    o.user_id,
    o.product_name,
    o.price,
    u.user_name,
    u.user_addr,
    -- Extract the postal code that follows the '#' separator in user_addr
    REGEXP_EXTRACT(u.user_addr, '#(.*?)$', 1) AS user_addr_code
FROM
    `lindorm_table`.`default`.`order_tbl` o
    JOIN `lindorm_table`.`default`.`user_tbl` u
    ON o.user_id = u.user_id;

Connect to LindormTable from the MySQL command line.

Insert sample data into both source tables.

INSERT INTO user_tbl (user_id, user_name, user_addr) VALUES
('U001', 'Zhang San', 'Chaoyang District, Beijing#100000'),
('U002', 'Li Si',     'Pudong New Area, Shanghai#200000'),
('U003', 'Wang Wu',   'Tianhe District, Guangzhou#510000'),
('U004', 'Zhao Liu',  'Nanshan District, Shenzhen#518000');

INSERT INTO order_tbl (order_id, user_id, product_name, price) VALUES
('O1001', 'U001', 'Laptop',               8999.00),
('O1002', 'U001', 'Wireless Mouse',         159.00),
('O1003', 'U002', 'Smartphone',            6999.00),
('O1004', 'U002', 'Bluetooth Headset',      299.00),
('O1005', 'U003', 'Tablet',               3499.00),
('O1006', 'U004', 'Mechanical Keyboard',    799.00),
('O1007', 'U004', 'Monitor',              1299.00);

Query the sink table. The ETL job writes the pre-joined results in real time as data arrives.

SELECT * FROM user_order_tbl;

Expected output:

+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+
| order_id | user_id | product_name        | price                     | user_name | user_addr                          | user_addr_code |
+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+
| O1001    | U001    | Laptop              | 8999.00000000000000000000 | Zhang San | Chaoyang District, Beijing#100000  | 100000         |
| O1002    | U001    | Wireless Mouse      |  159.00000000000000000000 | Zhang San | Chaoyang District, Beijing#100000  | 100000         |
| O1003    | U002    | Smartphone          | 6999.00000000000000000000 | Li Si     | Pudong New Area, Shanghai#200000   | 200000         |
| O1004    | U002    | Bluetooth Headset   |  299.00000000000000000000 | Li Si     | Pudong New Area, Shanghai#200000   | 200000         |
| O1005    | U003    | Tablet              | 3499.00000000000000000000 | Wang Wu   | Tianhe District, Guangzhou#510000  | 510000         |
| O1006    | U004    | Mechanical Keyboard |  799.00000000000000000000 | Zhao Liu  | Nanshan District, Shenzhen#518000  | 518000         |
| O1007    | U004    | Monitor             | 1299.00000000000000000000 | Zhao Liu  | Nanshan District, Shenzhen#518000  | 518000         |
+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+