背景

交易数据的实时统计是电商网站一个核心功能，可以帮助用户实时统计网站的整体销售情况，快速验证“新销售策略”的效果。我们今天介绍一个基于表格存储 (Tablestore) 实现交易数据的实时计算，给大家提供一个新使用方式。

Tablestore 作为在线的结构化数据库，提供了毫秒级的访问延时和丰富的查询方式，能高效的支撑交易数据的存储和查询，同时 Tablestore 已经原生支持阿里云的流计算框架 Flink/Blink，可以实现数据的实时计算。

架构

示例设计

基本场景

注意：示例是模拟一个电商网站的交易数据的存储和实时计算，目的是为了展示 Tablestore + Blink 的使用流程。

用户通过 SDK 将网站交易数据实时的存储 (PutRow/BatchWrite/TableStoreWriter) 到 Tablestore 的 source_order 表中，Tablestore 通过 Tunnel 服务，将实时增量的数据流入到 Flink/Blink 中，每 5 秒进行一次聚合计算，并将计算的结果重新写回 Tablestore 的 sink_order 表中。最后提供给“大屏”实时读取 (GetRange) 展示。

Source 表（源表）- source_order

source 表是原始数据表，存储了所有交易记录。

Sink 表（结果表）- sink_order

Flink SQL

DDL 参考

注意：当前 Blink 在支持 Tablestore source 上还有些限制，只能运行 ProcessingTime 的方式，未来会支持 EventTime 模式，按照用户数据参数的时间进行计算。

-- Source 源表创建
CREATE TABLE ots_input (
    metering VARCHAR,
    orderid VARCHAR,
    price DOUBLE,
    byerid BIGINT,
    sellerid BIGINT,
    productid BIGINT,
    primary key(metering,orderid),
    ts AS PROCTIME()) WITH (
    type = 'ots',
    instanceName = 'ordertest',
    tableName = 'source_order',
    accessId = '******************',
    accessKey = '******************',
    endpoint = 'http://ordertest.cn-zhangjiakou.vpc.tablestore.aliyuncs.com',
    tunnelName = 'blink_agg'
);
-- Sink 结果表创建
CREATE TABLE ots_output (
    metering VARCHAR,
    ts BIGINT,
    price DOUBLE,
    ordercount BIGINT,
    primary key(metering,ts)
) WITH (
    type = 'ots',
    instanceName = 'ordertest',
    tableName = 'sink_order',
    accessId = '******************',
    accessKey = '******************',
    endpoint = 'http://ordertest.cn-zhangjiakou.vpc.tablestore.aliyuncs.com',
    valueColumns = 'price,ordercount'
);

-- 计算
INSERT INTO ots_output
SELECT 
    DISTINCT metering as metering,
    CAST(TUMBLE_START(ots_input.ts, INTERVAL '5' SECOND) AS BIGINT) AS ts,
    SUM(price) as price,
    COUNT(orderid) as ordercount
FROM ots_input
GROUP BY TUMBLE(ts, INTERVAL '5' SECOND),metering;

实战

第一步：准备账户与开服

准备表格存储实例，可以参考《表格存储实例创建》

准备 Flink/Blink 项目，可以参考《Blink 如何购买》

第二步：创建资源

表格存储资源

表格存储控制台入口，创建表格存储实例 ordertest (用户自定义即可，下面对于参数位置更换为自定义的实例名)，并记录实例的 VPC 地址

同时在控制台创建 Source 表和 Sink 表，并为 Source 表（source_order）开启一个 Tunnel 服务 blink_agg

Blink 资源

Blink 控制台入口，创建一个 Blink 项目（独享模式要创建集群之后才能创建项目），分别创建一个作业，agg_order，并将上面的 Flink SQL 粘贴到窗口中，上线服务

在运维窗口中启动该任务

第三步：压入数据 并 实时获取结算结果

1 准备配置文件

程序默认会从 ’~/tablestoreConf.json’ 获取配置

vim ~/tablestoreConf.json

# 内容
{
    "endpoint":"http://ordertest.cn-zhangjiakou.ots.aliyuncs.com",
    "accessId":"************",
    "accessKey":"************",
    "instanceName":"ordertest"
}

2 构建源码

mvn install
cd target
tar xzvf stream-compute-1.0-SNAPSHOT-release.tar.gz

3 启动压力器和模拟大屏

可以直接下载工具包：stream-compute-1.0-SNAPSHOT-release.tar.gz

# 窗口 1
./bin/mock_order_generator
# 窗口 2
./bin/data_show_screen

4 效果

源码

源码：https://github.com/aliyun/tablestore-examples/tree/master/demos/StreamCompute

本文作者：chen1255

阅读原文

本文为云栖社区原创内容，未经允许不得转载。