开发者社区> 阿里云存储服务> 正文

Tablestore + Blink实战:交易数据的实时统计

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

背景

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

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

架构

image

示例设计

基本场景

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

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

Source表(源表)- source_order

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

字段 类型 注释
metering(PrimaryKey) string 计量类型,样例中默认是web
orderid(PrimaryKey) string 订单号ID
ts integer 交易时间(Unix时间戳,毫秒精度)
price double 交易金额
buyerid integer 买家ID
sellerid integer 卖家ID
productid integer 商品ID

Sink表(结果表)- sink_order

字段 类型 注释
metering(PrimaryKey) string 计量类型,样例中默认是web
ts(PrimaryKey) integer 交易时间(Unix时间戳,毫秒精度)
price double 交易金额
ordercount integer 交易次数

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地址

image

同时在控制台创建Source表和Sink表, 并为Source表(source_order)开启一个Tunnel服务blink_agg

image

图三 Source表(source_order)

image

图四 Sink表(sink_order)

image

图五 源表和目标表

image

图六 创建通道

Blink资源

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

image

image

image
在运维窗口中启动该任务
image

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

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 效果

image

源码

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

欢迎加入
如果您对表格存储使用有疑问、想探讨,欢迎加入【表格存储公开交流群】,群号:11789671。

image

版权声明:本文中所有内容均属于阿里云开发者社区所有,任何媒体、网站或个人未经阿里云开发者社区协议授权不得转载、链接、转贴或以其他方式复制发布/发表。申请授权请邮件developerteam@list.alibaba-inc.com,已获得阿里云开发者社区协议授权的媒体、网站,在转载使用时必须注明"稿件来源:阿里云开发者社区,原文作者姓名",违者本社区将依法追究责任。 如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件至:developer2020@service.aliyun.com 进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
上一篇:药品监管系统架构揭秘:海量溯源数据存储与查询