DPI エンジンのリアルタイム ETL - Lindorm - Alibaba Cloud ドキュメントセンター

Stream DPI エンジンは、テーブル単位の事前集計を実現するための抽出・変換・書き出し（ETL）SQL を提供し、リアルタイムで動作するマテリアライズドビューとして機能します。本ガイドでは、以下の 2 つのシナリオについて説明します：リアルタイムでのミラーテーブル同期、および複数テーブルの事前 JOIN によるクエリごとの JOIN オーバーヘッドの排除。

シナリオ 1：リアルタイムミラーテーブル
シナリオ 2：リアルタイム事前集計

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

Stream DPI エンジンを有効化済みであること
LindormTable を有効化済みであること

シナリオ 1：リアルタイムミラーテーブル

ミラーテーブルは、ソーステーブルへのすべての書き込みをリアルタイムで同期します。このパターンは、アプリケーション層の変更を伴わず、データ共有、データバックアップ、読み書き分離、異種インデックス構築などの用途に利用できます。

手順 1：テーブルの準備

MySQL コマンドラインから LindormTable に接続します。

ソーステーブルとミラーテーブルを作成します。

-- ソーステーブル
CREATE TABLE source(p1 INT, c1 DOUBLE, PRIMARY KEY(p1));

-- ミラー（シンク）テーブル — ソーステーブルと同じスキーマ
CREATE TABLE sink(p1 INT, c1 DOUBLE, PRIMARY KEY(p1));

手順 2：ETL ジョブの登録

MySQL コマンドラインからストリームエンジンに接続します。
ETL ジョブの作成を行い、source テーブルから sink テーブルへ全行をレプリケートします。
```
CREATE ETL sync_etl AS INSERT INTO sink SELECT * FROM source;
```

手順 3：結果の確認

MySQL コマンドラインから LindormTable に接続します。

ソーステーブルにレコードを挿入します。

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);

ミラーテーブルをクエリして、レコードが正しく同期されたことを確認します。

SELECT * FROM sink;

期待される出力：

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

シナリオ 2：リアルタイム事前集計

各クエリごとに JOIN を実行すると、パフォーマンス上のオーバーヘッドが発生します。事前 JOIN では、書き込みが到着するタイミングで複数テーブルのデータを一度だけマージし、その結果を単一のワイドテーブルに格納します。その後の読み取り処理では、事前 JOIN 済みのテーブルのみを参照するため、ランタイムでの JOIN は不要になります。

本シナリオでは、order_tbl と user_tbl を user_id で JOIN し、アドレスフィールドから正規表現を用いて郵便番号を抽出します。マージ後の結果は、リアルタイムで user_order_tbl に書き込まれます。

手順 1：テーブルの準備

MySQL コマンドラインから LindormTable に接続します。

2 つのソーステーブルと、order_tbl 上のセカンダリインデックスを作成します。

-- ソーステーブル 1：ユーザー情報
CREATE TABLE `user_tbl` (
    `user_id`   varchar NOT NULL,
    `user_name` varchar,
    `user_addr` varchar,
    PRIMARY KEY (`user_id`)
);

-- ソーステーブル 2：注文情報
CREATE TABLE `order_tbl` (
    `order_id`     varchar NOT NULL,
    `user_id`      varchar,
    `product_name` varchar,
    `price`        decimal(38, 20),
    PRIMARY KEY (`order_id`)
);

-- user_id での JOIN をサポートするための order_tbl のセカンダリインデックス。
-- このインデックスがない場合、ETL エンジンは user_tbl の書き込みごとに
-- order_tbl の全表スキャンを実行する必要があります。
CREATE INDEX idx1 ON `order_tbl`(user_id desc) WITH (COMPRESSION='ZSTD');

事前 JOIN 済みのデータを格納するシンクテーブルを作成します。

-- シンクテーブル：ユーザー情報と注文情報を統合し、郵便番号も抽出。
-- MUTABILITY='MUTABLE_UDT' を指定することで、片方のソーステーブルのみが更新された場合
-- （例：ユーザーが住所を変更した場合）にも部分カラム更新が可能になります。
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');

手順 2：ETL ジョブの登録

MySQL コマンドラインからストリームエンジンに接続します。

ETL ジョブの作成を、結合および変換ロジックを用いて行います。

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,
    -- 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;

手順 3：結果の確認

MySQL コマンドラインから LindormTable に接続します。

両方のソーステーブルにサンプルデータを挿入します。

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);

シンクテーブルをクエリします。ETL ジョブは、データ到着と同時にリアルタイムで事前 JOIN 済みの結果を書き込みます。

SELECT * FROM user_order_tbl;

期待される出力：

+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+
| 注文ID   | ユーザーID | プロダクト名        | 価格                      | ユーザー名 | ユーザー住所                         | ユーザー住所コード |
+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+
| 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         |
+----------+---------+---------------------+---------------------------+-----------+------------------------------------+----------------+