基于Tablestore的一站式物联网存储解决方案-Flink 实时计算-阿里云开发者社区

基于Tablestore的一站式物联网存储解决方案-Flink 实时计算

2021-09-02 599

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： ## 需求分析在共享充电宝场景中，会有一些对实时性要求较高的计算场景。例如大屏展示每个省份、每个机柜的营收情况。这类场景不同于离线计算类的场景，需要实时地根据订单数据的变化来统计营业额，并不能采用前文中介绍的Spark近实时流批计算来实现。整个场景需求链路为根据订单增量变化，触发计算逻辑，计算结果写入Tablestore表中，提供给大屏展示。大致可以归纳成**增量写入-实时

需求分析

在共享充电宝场景中，会有一些对实时性要求较高的计算场景。例如大屏展示每个省份、每个机柜的营收情况。这类场景不同于离线计算类的场景，需要实时地根据订单数据的变化来统计营业额，并不能采用前文中介绍的Spark近实时流批计算来实现。整个场景需求链路为根据订单增量变化，触发计算逻辑，计算结果写入Tablestore表中，提供给大屏展示。大致可以归纳成增量写入-实时计算-回写/展示一系列动作。

Tablestore通道服务介绍

在介绍实时计算实现方案之前，先需要了解一下表格存储Tablestore功能之一通道服务。
通道服务是表格存储数据接口上的全增量一体化服务。可提供增量、全量、全增量三种分布式数据实时消费通道。通过对数据表建立通道，可以实现对表中存量、增量数据进行实时消费。下图介绍了通道服务的一些常见应用场景

Tablestore+Blink方案

Tablestore实现了实时计算Blink的connector，可以基于通道服务实现实时消费增量数据。本方案共分四个模块。

订单源表。以表格存储Tablestore作为订单源表的存储库。在源表上建立增量类型通道，可实时监听表格中的数据变化。
Blink集群。Blink提供了分布式实时计算服务，可无缝动态对接表格存储通道服务，对增量数据进行分布式计算，并回写到订单结果表中。
订单结果表。以表格存储Tablestore作为订单数据计算完的结果表。
DataV。通过DataV访问表格存储，可将表格存储数据可视化展现。

通过上述四个模块之间的对接可实现此场景需求，方案整体架构如下图所示:

实现步骤

创建源表和结果表

登录实时计算官网。开通实时计算服务。
登录表格存储官网，进入控制台，进入订单表管理-实时消费通道页面。创建增量类型通道。

创建Blink独享集群。进入集群管理创建项目。

进入项目，编写作业SQL。
1. 创建订单数据源表

create table order_source(
  order_Md5ID VARCHAR,
  cabinet_province VARCHAR,
  order_start_time BIGINT  ,
  order_end_time BIGINT ,
  cabinet_pricePerHour  DOUBLE 
) with (
  type ='ots',
  endPoint ='Tablestore实例VPC服务地址',
  instanceName = 'Tablestore实例名',
  tableName ='order',
  tunnelName = '增量通道名',
  accessId ='',
  accessKey ='',
  ignoreDelete = 'false'
);

创建订单结算结果表

CREATE TABLE order_sink (
    order_province VARCHAR,
    total_price DOUBLE,
    order_count BIGINT,
    primary key(order_province)
) WITH (
    type = 'ots',
    endPoint ='Tablestore实例VPC服务地址',
      instanceName = 'Tablestore实例名',
      tableName ='order_sink',
      accessId ='',
      accessKey ='',
    valueColumns = 'total_price,order_count'
);

实时计算作业

编写实时计算作业SQL。按照省份实时聚合营业额，计算结果回写Tablestore结果表。

INSERT INTO order_sink
SELECT
    cabinet_province AS province,
    sum((order_end_time - order_start_time)/3600/1000 * cabinet_pricePerHour) AS             total_price,
    count(order_Md5ID) AS order_count
    FROM order_source
    GROUP BY cabinet_province;