Data Transmission Service (DTS) の抽出・変換・書き出し (ETL) 機能では、コードを記述せずに、視覚的でドラッグアンドドロップ可能な DAG インターフェイスを使用して、ストリーミングデータをリアルタイムで処理できます。本チュートリアルでは、リアルタイムのトランザクションデータとプロダクト参照データを結合し、価格しきい値を超える注文をフィルター処理して、結果をターゲットデータベースに書き込むという一連の操作を実践的に解説します。
適用範囲/利用シーン
集中型マルチソースデータ管理：複数のリージョンまたは異種のデータソースから取得したデータを、リアルタイムで単一のデータベースに統合します。
リアルタイムレポート作成：製品、顧客、時間などのさまざまな次元で、リアルタイムのビジネス活動を反映するレポートパイプラインを構築します。
リアルタイムコンピューティング：ストリーミングデータをクリーニングし、ユーザーのプロファイリング、リスク管理、レコメンデーションシステムなどに活用するための特徴量およびタグを抽出します。
仕組み
本例では、以下の 2 つのデータソースをマージします。
ストリームテーブル —
test_ordersテーブル。リアルタイムのトランザクションイベント（注文 ID、顧客 ID、プロダクト ID、合計金額、注文日付）を受信します。ディメンションテーブル —
productテーブル。比較的静的なプロダクト参照データ（プロダクト ID、プロダクト名、単位価格）を保持します。
ETL タスクは、これらの 2 つのテーブルを結合し、total_price > 3000.00 を満たす注文をフィルター処理したうえで、該当する行をリアルタイムでターゲットテーブルに書き込みます。
ストリームテーブルは、イベントが到着するごとに継続的に更新されます。一方、ディメンションテーブルは、変更頻度が低く、ストリームデータを豊かにするために使用される参照データを保持しており、ルックアップテーブルと同様の役割を果たします。
前提条件
開始する前に、以下の条件を満たしていることを確認してください。
DTS コンソール へのアクセス権限
ソースとして機能する ApsaraDB RDS for MySQL インスタンス（ストリームテーブルおよびディメンションテーブルを含む）
ターゲットとして機能する ApsaraDB RDS for MySQL インスタンス
ソースおよびターゲットテーブルの準備
ApsaraDB RDS for MySQL インスタンスに、以下のテーブルを作成します。
リアルタイムトランザクションデータ
CREATE TABLE test_orders (
order_id BIGINT NOT NULL COMMENT '注文 ID',
user_id BIGINT NOT NULL COMMENT 'ユーザー ID',
product_id BIGINT NOT NULL COMMENT 'プロダクト ID',
total_price DECIMAL(15,2) NOT NULL COMMENT '合計金額',
order_date TIMESTAMP NOT NULL COMMENT '注文日付',
PRIMARY KEY (order_id)
);ビジネスディメンションデータ
CREATE TABLE product (
product_id BIGINT NOT NULL COMMENT 'プロダクト ID',
product_name VARCHAR(20) COMMENT 'プロダクト名',
product_price DECIMAL(15,2) NOT NULL COMMENT '単位価格'
);ターゲットテーブル
CREATE TABLE test_orders_new (
order_id BIGINT NOT NULL COMMENT '注文 ID',
user_id BIGINT NOT NULL COMMENT 'ユーザー ID',
product_id BIGINT NOT NULL COMMENT 'プロダクト ID',
total_price DECIMAL(15,2) NOT NULL COMMENT '合計金額',
order_date TIMESTAMP NOT NULL COMMENT '注文日付',
product_id_2 BIGINT NOT NULL COMMENT 'プロダクト ID（product テーブルより）',
product_name VARCHAR(20) COMMENT 'プロダクト名',
product_price DECIMAL(15,2) NOT NULL COMMENT '単位価格',
PRIMARY KEY (order_id)
);ターゲットテーブルでは、product_id_2を使用して、productディメンションテーブルからのプロダクト ID を格納します。これにより、product_idという列名がtest_ordersストリームテーブルにも存在する場合の列名衝突を回避します。結合対象の 2 つのテーブルで同一の列名が使用されている場合は、ETL タスクの設定前に、ターゲットテーブルのスキーマにおいて片方の列名を変更してください。
ETL タスクの設定
以下の手順では、ストリームテーブルとディメンションテーブルを結合し、結果をフィルター処理してターゲットに書き込むデータフローを設定します。
ステップ 1：ソースデータベースの設定
左側ナビゲーションウィンドウで、ETL をクリックします。
ストリーミング ETL ページの左上隅にある
をクリックします。データフローの作成 ダイアログボックスで、データフロー名 フィールドに名前を入力し、開発方法 を DAG に設定します。OK をクリックします。
ストリームテーブルを設定します。
左側パネルから、入力／ディメンションテーブル MySQL ノードをキャンバスにドラッグします。
キャンバス上の 入力／ディメンションテーブル MySQL-1 をクリックします。
ノード設定 タブで、以下のパラメーターを設定します。
パラメーター 説明 データソース名 DTS が自動的に名前を生成します。識別が容易になるよう、説明的な名前を入力してください。 リージョン ソースデータベースのリージョンを選択します。対応するリージョン：中国 (杭州)、中国 (上海)、中国 (青島)、中国 (北京)、中国 (張家口)、中国 (深セン)、中国 (広州)、中国 (香港)。 インスタンス ソースデータベースインスタンスを選択します。新しいインスタンスを作成するには、インスタンスの作成 をクリックします。詳細については、「DMS でサポートされるデータベース」をご参照ください。 ノードタイプ ストリームテーブル を選択します。ストリームテーブルは継続的に更新され、ディメンションテーブルと結合できます。 フォーマット変換 動的テーブルをストリームに書き戻す際のエンコーディング方式を指定します。アップサートストリーム：INSERT および UPDATE 操作はアップサートメッセージとして、DELETE 操作は削除メッセージとしてエンコードされます。一意キー（複合キーも可）が必要です。追記専用ストリーム：出力ストリームには INSERT 操作のみが含まれます。 データベースおよびテーブルの選択 変換対象のデータベースおよびテーブルを選択します。 出力フィールド タブで、含める列を選択します。
時刻属性
パラメーター 説明 イベント時刻ウォーターマーク 各イベントが生成された時刻を表すタイムスタンプフィールドを選択します（例： order_date）。イベント時刻ウォーターマークの遅延 許容される最大の順序外到着遅延時間を入力します。ETL は、この時間だけ遅れて到着するイベントを待機したうえで、破棄します。たとえば、9:59 に生成されたデータが、10:00 に加えて設定された遅延時間が経過しても到着しない場合、そのデータは破棄されます。 処理時間 列名を入力します。ETL はサーバーの処理時間をこの列に格納します。最新バージョンのディメンションテーブルを常に参照するテンポラル結合には、処理時間を使用します。
ノードの右側に
アイコンが表示されなくなった場合、ストリームテーブルの設定は完了しています。ディメンションテーブルを設定します。
左側パネルから、もう 1 つの 入力／ディメンションテーブル MySQL ノードをキャンバスにドラッグします。
キャンバス上の 入力／ディメンションテーブル MySQL-2 をクリックします。
ノード設定 タブで、以下のパラメーターを設定します。
パラメーター 説明 データソース名 DTS が自動的に名前を生成します。識別が容易になるよう、説明的な名前を入力してください。 リージョン ソースデータベースのリージョンを選択します。 インスタンス ソースデータベースインスタンスを選択するか、インスタンスの作成 をクリックします。 ノードタイプ ディメンションテーブル を選択します。ディメンションテーブルは、変更頻度が低く、ストリームデータをワイドテーブルに拡張するために使用される参照データを保持します。 データベースおよびテーブルの選択 変換対象のデータベースおよびテーブルを選択します。 出力フィールド タブで、含める列を選択します。
ノードの右側に
アイコンが表示されなくなった場合、ディメンションテーブルの設定は完了しています。
ステップ 2：テーブル結合コンポーネントの設定
左側パネルの 変換 セクションから、JOIN ノードをキャンバスにドラッグします。
ストリームテーブルノードを [Table Join-1] に接続します。ストリームテーブルノードの上にカーソルを合わせ、その右端にある中抜き円をクリックして、[Table Join-1] まで接続線をドラッグします。ディメンションテーブルノードについても同様に繰り返します。
キャンバス上の テーブル結合-1 をクリックして、設定を開きます。
ノード設定 タブで、以下のパラメーターを設定します。
セクション パラメーター 説明 変換名 変換名の入力 DTS が自動的に名前を生成します。説明的な名前を入力してください。 JOIN 設定 JOIN 句における左側テーブル 主テーブル（左側テーブル）としてストリームテーブルを選択します。 テンポラル結合時刻属性 ストリームテーブルのバージョンとディメンションテーブルのバージョンをどのように照合するかを定義します。未設定の場合、通常の結合が実行されます。イベント時刻ウォーターマークに基づく：イベントが生成された時点でのディメンションテーブルのバージョンと照合します（バージョン管理テーブルで使用）。処理時間に基づく：常にディメンションテーブルの最新バージョンと結合します（標準テーブルで使用）。本例では、処理時間に基づく を選択します。 結合操作の選択 結合タイプを選択します。本例では、内部結合 を選択します。内部結合：両テーブルに一致する値を持つ行のみを返します。左結合：ストリームテーブルのすべての行と、利用可能な場合のディメンションテーブルの一致する行を返します。右結合：ディメンションテーブルのすべての行と、利用可能な場合のストリームテーブルの一致する行を返します。 結合条件 + 条件の追加 + 条件の追加 をクリックし、結合対象のフィールドを選択します。 =記号の左側のフィールドは左側テーブルに属し、右側のフィールドは右側テーブルに属します。出力フィールド タブで、ワイドテーブルに含める列を選択します。
テーブル結合-1 コンポーネントの右側に
アイコンが表示されなくなった場合、結合の設定は完了しています。
ステップ 3：テーブルレコードフィルター コンポーネントの設定
左側パネルの 変換 セクションから、テーブルレコードフィルター ノードをキャンバスにドラッグします。
テーブル結合-1 を テーブルレコードフィルター-1 に接続します：テーブル結合-1 の上にカーソルを合わせ、右端の中抜き円をクリックして、テーブルレコードフィルター-1 まで接続線をドラッグします。
キャンバス上の テーブルレコードフィルター-1 をクリックします。
ノード設定 タブで、変換名 フィールドに名前を入力します。
WHERE 条件 フィールドで、以下のいずれかの方法でフィルター条件を指定します。
条件を直接入力します。たとえば、合計金額が 3,000 を超える注文のみを通過させるには、
total_price>3000.00を入力します。入力フィールド セクションのフィールドおよび 演算子 セクションの演算子をクリックして、視覚的に条件を構築します。
コンポーネントの右側に
アイコンが表示されなくなった場合、フィルターの設定は完了しています。
ステップ 4：ターゲットデータベースの設定
左側パネルの 出力 セクションから、MySQL ノードをキャンバスにドラッグします。
テーブルレコードフィルター-1 を 出力 MySQL-1 に接続します：テーブルレコードフィルター-1 の上にカーソルを合わせ、右端の中抜き円をクリックして、出力 MySQL-1 まで接続線をドラッグします。
キャンバス上の 出力 MySQL-1 をクリックします。
ノード設定 タブで、以下のパラメーターを設定します。
パラメーター 説明 データソース名 DTS が自動的に名前を生成します。説明的な名前を入力してください。 リージョン ターゲットデータベースのリージョンを選択します。対応するリージョン：中国 (杭州)、中国 (上海)、中国 (青島)、中国 (北京)、中国 (張家口)、中国 (深セン)、中国 (広州)、中国 (香港)。 インスタンス ターゲットデータベースインスタンスを選択するか、インスタンスの作成 をクリックします。 テーブルマッピング ターゲットテーブルの選択 セクションで、ターゲットテーブルをクリックします。 出力フィールド タブで、ターゲットテーブルに書き込む列を選択します。
コンポーネントの右側に
アイコンが表示されなくなった場合、ターゲットデータベースの設定は完了しています。
ステップ 5：事前チェックおよびタスクの開始
Flink SQL 検証の生成 をクリックします。DTS は Flink SQL ステートメントを生成し、検証チェックを実行します。
検証が完了したら、ETL 検証の詳細の表示 をクリックして、生成された SQL およびチェック結果を確認します。確認後、閉じる をクリックします。
検証に失敗した場合は、結果にリストされた問題を修正してから、再度 Flink SQL を生成してください。
次へ：タスク設定の保存および事前チェック をクリックします。事前チェックが成功した場合にのみ、タスクを開始できます。いずれかの項目が失敗した場合は、失敗した項目の横にある 詳細の表示 をクリックして問題を修正し、再度事前チェックを実行してください。
事前チェックが成功した後、次へ：インスタンスの購入 をクリックします。
インスタンスの購入 ページで、インスタンスクラス および コンピューティングユニット (CU) のパラメーターを設定します。Data Transmission Service（従量課金）サービス利用規約 および パブリックプレビュー向けサービス利用規約 を読み、同意してください。
パブリックプレビュー期間中は、各ユーザーが最大 2 つの ETL インスタンスを無料で作成できます。
購入および開始 をクリックして、ETL タスクを起動します。
結果
タスクが開始されると、DTS は test_orders から変更イベントを継続的に読み取り、product ディメンションテーブルと結合し、total_price > 3000.00 のフィルター条件を適用して、該当する行をリアルタイムで test_orders_new に書き込みます。
図 1．リアルタイムトランザクションデータテーブル：test_orders
図 2．ターゲットテーブル：test_orders_new
次に何をすべきか
DAG モードでの ETL タスクの設定 — すべての ETL タスク設定パラメーターの完全なリファレンス
DMS でサポートされるデータベース — サポートされるソースおよびターゲットデータベースの種類