E-MapReduce:ホットデータとコールドデータの階層化ストレージ

前提条件

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

• Streaming SQL が有効化された EMR クラスター

• Tablestore インスタンスおよびテーブルとトンネルを作成する権限

• サンプルデータを書き込むための Tablestore SDK for Java

ステップ 1：ソーステーブルの設定とサンプルデータのロード

プライマリキー列 (UserId および OrderId) と属性列 (price および timestamp) を持つ、OrderSource という名前の Tablestore テーブルを作成します。

Tablestore SDK for Java の BatchWriteRow 操作を使用して、注文データをテーブルに書き込みます。本例では、タイムスタンプ範囲は直近 90 日間 (2020-02-26 ～ 2020-05-26) をカバーし、レコード総数は 3,112,400 件です。

データ書き込み時に、timestamp 列を Tablestore 行のバージョン番号属性列として設定します。テーブルに TTL を設定し、保存期間を超えたバージョン番号を持つ行が自動的に削除されるようにします。

ステップ 2：CDC トンネルの作成

Tablestore コンソールで、OrderSource テーブル上に増分同期用のトンネルを作成します。トンネル ID を記録しておき、Streaming SQL の構成で使用します。

ステップ 3：Streaming SQL の起動とソース・シンクテーブルの作成

ご利用の EMR クラスターの Header ノードで、Streaming SQL インタラクティブシェルを起動します。

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

以下のコマンドを実行して、Tablestore ソーステーブル、Delta Lake シンク、増分スキャンビュー、およびストリーミングジョブを作成します。

-- 1. Tablestore ソーステーブルの作成。
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. Delta Lake シンクの作成。
DROP TABLE IF EXISTS delta_orders;
CREATE TABLE delta_orders(
  UserId string,
  OrderId string,
  price double,
  timestamp long
)
USING delta
LOCATION '/delta/orders';

-- 3. ソーステーブルの増分スキャンビューの作成。
--    tunnel.id: ステップ 2 で作成したトンネルの ID。
--    maxoffsetsperchannel: トンネル経由で各パーティションに書き込める最大データ量。
CREATE SCAN incremental_orders ON order_source USING STREAM
OPTIONS(
  tunnel.id="324c6bee-b10d-4265-9858-b829a1b71b4b",
  maxoffsetsperchannel="10000"
);

-- 4. ストリーミングジョブの開始。
--    このジョブは CDC 変更 (PUT / UPDATE / DELETE) を delta_orders にマージします。
--    __ots_record_type__ は Tablestore ストリーミングソースから提供される事前定義列で、
--    行操作タイプを識別します。
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);

上記 4 つの SQL ステートメントは、それぞれ以下の処理を行います。

ステップ 4：初期同期の検証と TTL の設定

TTL を変更する前に、両方のテーブルに同じ件数のレコードが含まれていることを確認します。

order_source および delta_orders のレコード件数をクエリします。件数は一致している必要があります (各 3,112,400 件)。

OrderSource テーブルの TTL を 30 日に設定します。TTL が適用されると、以下のようになります。

• order_source は直近 30 日分のレコードのみを保持します：1,017,004 件。

• delta_orders は引き続き全レコードを保持します：3,112,400 件。

これにより、ホットデータとコールドデータが分離されました。