Tablestore:ワイドカラムモデルのチュートリアル

DDL 構文

次のサンプルコードは、ソーステーブルの DDL 構文の例を示しています。

CREATE TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector'='ots',
    'endPoint' ='https://flink-source.cn-hangzhou.ots.aliyuncs.com',
    'instanceName' = 'flink-source',
    'tableName' ='flink_source_table',
    'tunnelName' = 'flinksourcestream',
    'accessId' ='xxxxxxxxxxx',
    'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false' // 削除操作を無視するかどうかを指定します。
);

Tunnel Service は、消費されていないユーザーデータと OtsRecordType および OtsRecordTimestamp フィールドを返し、Realtime Compute for Apache Flink がデータとフィールドを属性列として読み取れるようにします。次の表に、フィールドを示します。

Realtime Compute for Apache Flink に OtsRecordType および OtsRecordTimestamp フィールドを読み込ませる場合は、Realtime Compute for Apache Flink が提供する METADATA キーワードを使用して、Tablestore ソーステーブルから属性フィールドを取得できます。次の例は、DDL ステートメントを示しています。

CREATE TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    record_type STRING METADATA FROM 'type',
    record_timestamp BIGINT METADATA FROM 'timestamp'
) WITH (
    ...
);

WITH 句のパラメータ

DDL 構文

次のサンプルコードは、結果テーブルの DDL 構文の例を示しています。

CREATE TABLE ots_sink (
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    PRIMARY KEY (`order`,orderid) NOT ENFORCED
) WITH (
     ...
);

WITH 句のパラメータ

Tablestore と Realtime Compute for Apache Flink 間のフィールドのデータ型マッピング

前提条件

• AccessKey ペアが作成されていること。詳細については、AccessKey ペアの作成 を参照してください。

• Tablestore ソーステーブルのトンネルが作成されていること。トンネルの作成方法については、トンネルの作成 を参照してください。

ステップ 1: SQL ドラフトを作成する

1. Realtime Compute for Apache Flink コンソール にログインします。

2. フルマネージド Flink タブで、管理するワークスペースを見つけ、アクション 列の コンソール をクリックします。

3. 左側のナビゲーションペインで、SQL エディタ をクリックします。

4. SQL エディタページの左上隅にある 新規 をクリックします。

5. 新規ドラフト ダイアログボックスで、空のストリームドラフト をクリックします。

   フルマネージド Flink は、さまざまなコードテンプレートとデータ同期テンプレートを提供します。各コードテンプレートは、特定のシナリオ、コードサンプル、および手順を提供します。テンプレートをクリックすると、Realtime Compute for Apache Flink の機能と関連構文について学習し、ビジネスロジックを実装できます。詳細については、コードテンプレート と データ同期テンプレート を参照してください。

6. 次へ をクリックします。

7. ドラフトのパラメータを設定します。次の表に、パラメータを示します。

   パラメータ	説明	例
   名前	作成するドラフトの名前。 説明 ドラフト名は、現在のプロジェクト内で一意である必要があります。	flink-test
   場所	ドラフトのコードファイルが格納されるフォルダ。 既存のフォルダの右側にある アイコンをクリックして、サブフォルダを作成することもできます。	開発
   エンジンバージョン	ドラフトで使用される Flink のエンジンバージョン。エンジンバージョン、バージョンマッピング、各バージョンのライフサイクルにおける重要な時点の詳細については、エンジンバージョン を参照してください。	vvr-6.0.4-flink-1.15

8. 作成 をクリックします。

ステップ 2: ドラフトのコードを記述する

1. 一時ソーステーブルと一時結果テーブルを作成します。

   説明

   ドラフトを作成する際は、一時テーブルの使用回数を最小限に抑えることをお勧めします。また、カタログに登録されているテーブルを使用することをお勧めします。

   次のサンプルコードは、tablestore_stream という名前の一時ソーステーブルと ots_sink という名前の一時結果テーブルを作成する方法の例を示しています。

   CREATE TEMPORARY TABLE tablestore_stream(
       `order` VARCHAR,
       orderid VARCHAR,
       customerid VARCHAR,
       customername VARCHAR
   ) WITH (
       'connector'='ots',
       'endPoint' ='https://flink-source.cn-hangzhou.ots.aliyuncs.com',
       'instanceName' = 'flink-source',
       'tableName' ='flink_source_table',
       'tunnelName' = 'flinksourcestream',
       'accessId' ='xxxxxxxxxxx',
       'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
       'ignoreDelete' = 'false' // 削除操作を無視するかどうかを指定します。
   );
   
   CREATE TEMPORARY TABLE ots_sink (
       `order` VARCHAR,
       orderid VARCHAR,
       customerid VARCHAR,
       customername VARCHAR,
       PRIMARY KEY (`order`,orderid) NOT ENFORCED
   ) WITH (
       'connector'='ots',
       'endPoint'='https://flink-sink.cn-hangzhou.ots.aliyuncs.com',
       'instanceName'='flink-sink',
       'tableName'='flink_sink_table',
       'accessId'='xxxxxxxxxxx',
       'accessKey'='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
       'valueColumns'='customerid,customername'
   );

2. ドラフトロジックを記述します。

   次のサンプルコードは、ソーステーブルから結果テーブルにデータを挿入する方法の例を示しています。

   INSERT INTO ots_sink
   SELECT `order`, orderid, customerid, customername FROM tablestore_stream;

ステップ 3: [構成] タブでパラメータを設定する

SQL エディタページの右側にある 構成 タブをクリックし、次のパラメータを設定します。

• エンジンバージョン: Flink エンジンのバージョン。ドラフトの作成時に選択したバージョンを変更できます。

  重要

  VVR 3.0.3 以降のバージョンでは、Ververica Platform (VVP) を使用して、異なるエンジンバージョンを使用する SQL ジョブを同時に実行できます。VVR 3.0.3 を使用する Flink エンジンのバージョンは Flink 1.12 です。ジョブのエンジンバージョンが Flink 1.12 以前の場合は、ジョブで使用されているエンジンバージョンに基づいて、次の操作を実行してエンジンバージョンを更新できます。

  • Flink 1.12: ジョブを停止してから再起動します。その後、システムはジョブのエンジンバージョンを vvr-3.0.3-flink-1.12 に自動的に更新します。

  • Flink 1.11 または Flink 1.10: ジョブのエンジンバージョンを vvr-3.0.3-flink-1.12 または vvr-4.0.8-flink-1.13 に手動で更新してから、ジョブを再起動します。そうしないと、ジョブの開始時にタイムアウトエラーが発生します。

• 追加の依存関係: ドラフトで使用される追加の依存関係 (一時関数など)。

(オプション) ステップ 5: ドラフトをデバッグする

SQL エディタページの右上隅にある デバッグ をクリックします。

デバッグ機能を有効にして、デプロイ実行をシミュレートし、出力をチェックし、SELECT ステートメントと INSERT ステートメントのビジネスロジックを検証できます。この機能により、開発効率が向上し、データ品質低下のリスクが軽減されます。詳細については、ドラフトのデバッグ を参照してください。

ステップ 6: ドラフトをデプロイする

SQL エディタページの右上隅にある デプロイ をクリックします。ドラフトのデプロイ ダイアログボックスで、パラメータを設定し、確認 をクリックします。

ステップ 7: ドラフトのデプロイを開始し、計算結果を表示する

1. 左側のナビゲーションペインで、デプロイ をクリックします。

2. 開始するデプロイを見つけ、アクション 列の 開始 をクリックします。

   デプロイ起動パラメータの設定方法の詳細については、デプロイの開始 を参照してください。[開始] をクリックすると、デプロイステータスが 実行中 に変わります。これは、デプロイが想定どおりに実行されていることを示します。

3. [デプロイ] ページで、計算結果を表示します。

   1. 左側のナビゲーションペインで、デプロイ をクリックします。[デプロイ] ページで、管理するデプロイの名前をクリックします。

   2. 診断 タブをクリックします。

   3. ログ タブで、実行中のタスクマネージャ をクリックし、パス、ID 列の値をクリックします。

   4. ログ をクリックします。[ログ] タブで、シンクに関連するログを検索します。