使用 Flink SQL 访问 Tablestore 源表

2021-08-05 06:28:03

本文将介绍如何使用使用 Flink SQL 通过流处理的方式访问 Tablestore 源表。
在流计算场景下，用户可以基于通道服务，利用CDC（数据变更捕获）技术，通过 Flink 完成流式消费和计算。 Flink on Tablestore 可以提供 at-least-once 一致性语义。

Tablestore 简介

Tablestore 启发自 Google 的 Bigtable 论文，从 2009 年开始，在阿里云的飞天团队内，开始孵化。经过 10 年的锤炼，如今在集团、公有云和专有云积累了各式各样的客户和场景。
Tablestore 是一款 Serverless 云原生存储引擎，Serverless 相比实例售卖类型的产品，在业务有波峰波谷时天生就有较大的优势，基于 Bigtable 的主存储采用行的方式进行存储，可以支撑单表亿级别的 QPS。下面列了一些 Tablestore 的核心特性：

前置准备

由于本文是 Flink SQL - Tablestore 使用系列的第一篇文章，将会有一些篇幅介绍Flink实时计算平台的申请与使用。以下是几个前置准备步骤：

1. 购买 Flink 服务，步骤参见实时计算Flink版-产品文档，建议购买 Blink 独享集群（Flink 全托管 / 半托管暂不支持 Tablestore 源表）。如下图所示：

2. 创建Flink集群以及项目，如下图所示：

快速开始

步骤一创建 Tablestore 源表和数据通道

1. 创建 Tablestore 源表

Tablestore 的详细开通步骤请参考官方文档，本文demo中所创建出来的表名为 Demo, 其中，CellNumber、StartTime 作为数据表的主键，分别代表主叫号码和通话发生时间，CalledNumber、Duration 和 BaseStationNumber 三列为数据表的预定义列，分别代表被叫号码、通话时长和基站号码，数据示例如下图所示。

2. 在主表上创建数据通道，如下图所示，通道列表里面会显示该通道的名字、ID 以及类型。其中，通道名会用于后续的流式处理。

步骤二编写 Flink SQL 任务并启动

1. 进入 Flink 项目开发平台，进行任务开发，如下图所示：

2. 编写流任务

# 创建tablestore源表
CREATE TABLE Demo (
    CellNumber              VARCHAR,
    StartTime               BIGINT,
    BaseStationNumber       VARCHAR,
    CalledNumber            VARCHAR,
    Duration                BIGINT,
    OtsRecordType           VARCHAR header,
    OtsRecordTimestamp      BIGINT header,
    Duration_OtsColumnType  VARCHAR header
) WITH (
    type ='ots',
    endPoint ='https://instance_name.cn-hangzhou.vpc.tablestore.aliyuncs.com',
    instanceName = 'instance_name',
    tableName ='table_name',
    tunnelName = 'tunnel_name',
    accessId ='access_xxxxxx_id',
    accessKey ='access_xxxxxx_key',
    ignoreDelete = 'true'
);

# 创建print结果表（打印到taskmanager.out中）
CREATE TABLE print_sink (
    CellNumber              VARCHAR,
    StartTime               BIGINT,
    BaseStationNumber       VARCHAR,
    CalledNumber            VARCHAR,
    Duration                BIGINT,
    OtsRecordType           VARCHAR,
    OtsRecordTimestamp      BIGINT,
    Duration_OtsColumnType  VARCHAR
) WITH (
  type='print'
);

# 创建数据转换过程
INSERT INTO print_sink
SELECT *
FROM Demo;

源表字段header配置：

    - OtsRecordType：数据操作类型，包括PUT, UPDATE, DELETE。
    - OtsRecordTimestamp：数据操作时间（全量数据时，OtsRecordTimestamp为0）。
    - <列名>_OtsColumnType：某列的操作类型，包括PUT, UPDATE, DELETE。

源表参数配置：

    - type：数据源类型，tablestore源表为ots。
    - endPoint: 表格存储实例访问地址，建议用VPC地址。
    - instanceName：实例名。
    - tableName：表名
    - tunnelName：tunnel名。
    - accessId: 阿里云账号AK ID。
    - accessKey：阿里云账号AK Secret。
    - ignoreDelete: 是否忽略delete操作，默认值为false。

更详细的配置请参考：Flink官方文档-创建表格存储Tablestore源表

3. 点击保存后，上线。点击运维，查看流计算任务运行情况。具体如下图所示：