使用教程（宽表模型）

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' //是否忽略delete操作的数据.
);

除了待消费的用户数据外，Tunnel Service返回数据中的OtsRecordType、OtsRecordTimestamp字段均支持通过属性字段的方式读取。字段说明请参见下表。

当需要读取OtsRecordType和OtsRecordTimestamp字段时，Flink提供了METADATA关键字用于获取源表中的属性字段，具体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参数

结果表字段类型映射

前提条件

• 已创建AccessKey。具体操作，请参见创建AccessKey。

• 已为表格存储数据表（源表）创建数据通道。具体操作，请参见创建数据通道。

步骤一：创建作业

1. 登录实时计算控制台。

2. 在Flink全托管页签，单击目标工作空间操作列下的控制台。

3. 在左侧导航栏，单击SQL开发。

4. 单击新建。

5. 在新建作业草稿对话框中，单击空白的流作业草稿。

   Flink全托管也为您提供了丰富的代码模板和数据同步，每种代码模板都为您提供了具体的使用场景、代码示例和使用指导。您可以直接单击对应的模板快速地了解Flink产品功能和相关语法，实现您的业务逻辑，详情请参见代码模板和数据同步模板。

6. 单击下一步。

7. 填写作业配置信息。

   作业参数	说明	示例
   文件名称	作业的名称。 说明 作业名称在当前项目中必须保持唯一。	flink-test
   存储位置	指定该作业的代码文件所属的文件夹。 您还可以在现有文件夹右侧，单击图标，新建子文件夹。	作业草稿
   引擎版本	当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍。	vvr-6.0.4-flink-1.15

8. 单击创建。

步骤二：编写作业代码

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' //是否忽略delete操作的数据。
   );
   
   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;

步骤三：进行更多配置

在作业开发页面右侧，单击更多配置后，您可以填写以下参数信息：

• 引擎版本：修改您创建作业时选择的Flink引擎版本。

  说明 从VVR 3.0.3版本（对应Flink 1.12版本）开始，VVP支持同时运行多个不同引擎版本的SQL作业。如果您的作业已使用了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.7-flink-1.13版本后重启作业，否则会在启动作业时超时报错。

• 附加依赖文件：作业中需要使用到的附加依赖，例如临时函数等。

步骤五：（可选）进行作业调试

在作业开发页面顶部，单击调试。

您可以使用作业调试功能模拟作业运行、检查输出结果，验证SELECT或INSERT业务逻辑的正确性，提升开发效率，降低数据质量风险。具体操作，请参见作业调试。

步骤六：作业部署

在作业开发页面顶部，单击部署，在部署新版本对话框，可根据需要填写或选中相关内容，单击确认。

步骤七：启动并查看Flink计算结果

1. 在左侧导航栏，单击作业运维。

2. 单击目标作业名称操作列中的启动。

   作业启动参数配置详情请参见作业启动。单击启动后，您可以看到作业状态变为运行中，则代表作业运行正常。

3. 在作业运维详情页面，查看Flink计算结果。

   1. 在作业运维页面，单击目标作业名称。

   2. 单击作业探查。

   3. 在运行日志页签，单击运行Task Managers页签下的Path, ID。

   4. 单击日志，在页面搜索Sink相关的日志信息。