Tablestore:Tutorial (Wide Column model)

Tablestore source tables

In Realtime Compute for Apache Flink, you can use DDL statements of the source table to read the primary key and attribute column values of the source table in Tablestore based on the data type mapping of fields in Tablestore and Realtime Compute for Apache Flink. For more information, see Tablestore connector.

DDL syntax

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' // Specify whether to ignore delete operations.
);

Tunnel Service returns user data that was not consumed and the OtsRecordType and OtsRecordTimestamp fields, and allows Realtime Compute for Apache Flink to read the data and the fields as attribute columns. The following table describes the fields.

If you want Realtime Compute for Apache Flink to read the OtsRecordType and OtsRecordTimestamp fields, you can use the METADATA keyword that is provided by Realtime Compute for Apache Flink to obtain the attribute fields from the Tablestore source table. The following sample code provides an example of the DDL statement:

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

Parameters in the WITH clause

Data type mappings

Tablestore result tables

Realtime Compute for Apache Flink allows you to store results in a Tablestore table. For more information, see Tablestore connector.

DDL syntax

The following sample code provides an example of the DDL syntax of a result table.

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

Parameters in the WITH clause

Sample SQL statements

Read data from a source table

The following sample SQL statement provides an example on how to read data from a source table in batches:

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' // Specify whether to ignore delete operations. 
);
SELECT * FROM tablestore_stream LIMIT 100;

Synchronize data to a result table

The following sample SQL statement provides an example on how to write result data to the result table by calling the updateRow operation:

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' // Specify whether to ignore delete operations. 
);

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

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

Develop a real-time computing draft

Prerequisites

• An AccessKey pair is created. For more information, see Create an AccessKey pair.

• A role is assigned to the Alibaba Cloud account or the required permissions are granted to the RAM user by using an Alibaba Cloud account. For more information, see Assign a role to an Alibaba Cloud account or Grant permissions to a RAM user.

• A tunnel that uses the Tablestore data table as the source table is created. For more information, see Quick start.

Step 1: Create an SQL draft

1. Log on to the Realtime Compute for Apache Flink console.

2. On the Fully Managed Flink tab, find the workspace that you want to manage and click Console in the Actions column.

3. In the left-side navigation pane, click SQL Editor.

4. In the upper-left corner of the SQL Editor page, click New.

5. On the SQL Scripts tab of the New Draft dialog box, click Blank Stream Draft.

   Fully managed Flink provides various code templates and supports data synchronization. Each code template is applicable to specific scenarios and provides code samples and instructions for you. You can click a template to learn about the features and related syntax of Flink and implement your business logic. For more information, see Code templates and Data synchronization templates.

6. Click Next.

7. In the New Draft dialog box, configure the parameters of the draft. The following table describes the parameters.

   Parameter	Description	Example
   Name	The name of the draft that you want to create. Note The draft name must be unique in the current project.	flink-test
   Location	The folder in which the code file of the draft is saved. You can also click the icon to the right of an existing folder to create a subfolder.	Development
   Engine Version	The engine version of Flink that is used by the draft. For more information about engine versions, version mappings, and important time points in the lifecycle of each version, see Engine version.	vvr-6.0.4-flink-1.15

8. Click Create.

Step 2: Write code for the draft

1. Create a temporary source table and a temporary result table.

   Note

   When you create an SQL draft, we recommend that you minimize the number of times that temporary tables are used. We also recommend that you use tables that are registered in catalogs.

   The following sample code provides an example on how to create a tablestore_stream temporary table and an ots_sink temporary table:

   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' // Specify whether to ignore delete operations. 
   );
   
   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. Write the draft logic.

   The following sample code provides an example on how to select data from the source table and insert the data into the result table:

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

Step 3: Configure parameters on the Configurations tab