303      阿里云  技术社区[云栖]

阿里流计算平台开发实例之电商双11实时计算

  由于之前没写过博客之类的文章，所以这次写也是心中踹踹，也是由于这个项目间没有找到相关的一些文档，当时就想着完成后写一个出来，如果有写的不周到的地方，请联系我改正，谢谢。

一、 项目案例
用户商业模式含盖电商零售与加盟店批发零售，本次主要业务需求在于淘宝双11期间能实时计算用户所关注的一些指标数据，如：订单数、订单金额、商品SKU数、订单来源地、商品排名等等。
基于这些指标需求，除了要达到实时的要求以外，还需要具备适当的展现图设计，本次使用的是阿里云的DATAV，提供饼状图占比分析、商品与类目数据排名、国家地图热力展示等等。
二、 技术架构
由于用户的数据在云下，我们考虑的首先是迁移数据上云，再通过DTS将数据同步至DATAHUB，接着使用阿里流计算开发平台接入DATAHUB数据，并开发流计算代码，将执行结果输出至RDS MYSQL，最后DATAV引用RDS数据并开发图形展现界面。最终设计的技术架构如下图所示：

图：流计算数据逻辑设计图

三、 技术实现
1、 数据迁移与数据同步：由于数据不能直接到DataHub，使用阿里云DTS工具先完成数据迁移至RDS，链接：https://dts.console.aliyun.com/ 。再使用其数据同步功能，将RDS数据同步至DataHub(注：RDS收费可包月、DTS收费按小时)。在数据同步环节需要注意，根据企业数据量的大小，调整数据传输的通道大小。另外DataHub会自动创建对应同步的表的Topic，所以不需要在同步前自建Topic，建了会报错。（注意系统生成的Topic与自建的有哪些不同）
2、 StreamComputer流计算开发：其开发方式和技术要求，相比传统的开源产品，要简单许多，而且流计算平台功能比较丰富，特别是监控系统。其链接地址是：https://stream.console.aliyun.com。
2．1、datahub业务表的引用：
create table table_name( -- a
dts_column1 varchar, -- b
dts_column2 varchar,
dts_column3 bigint,
dts_column4 double,
dts_operation_flag varchar, -- c
dts_instance_id varchar,
dts_db_name varchar,
dts_table_name varchar,
dts_utc_timestamp varchar,
dts_before_flag varchar,
dts_after_flag varchar
) with (
type='datahub', -- d
endPoint='https://dh-cn-hangzhou.aliyuncs.com', -- e
project='project_name', -- f
topic='topic_name', -- g
accessId='accessId',
accessKey='accessKey',
startTime='2017-11-08 00:00:00' -- h
);
注解说明：
a、 在流计算引擎中建一张表，该表的名称是什么，建议和DataHub上一致
b、 要引用到该表的哪些字段，建议不需要的字段不要引用
c、 系统自建的TOPIC，该字段纪录的是该行数据是更新还是插入还是删除
d、 流计算可以引用多种数据源，这里表明数据源类型
e、 固定写法
f、 DataHub上的项目名称
g、 DataHub上的topic名称
h、 DataHub默认保留三天内的业务数据，该时间指定流计算引擎从哪个时间点取数
2.2、维表的引用：
create table div_product(
skucode varchar,
skuname varchar,
primary key(skucode), -- a
PERIOD FOR SYSTEM_TIME -- b
) with (
type='rds',
url='jdbc:mysql://yourHostName:3306/databaseName',
tableName=div_project,
userName='username',
password='password'
);
注解说明：
a、 该表的主键是什么，需要指定
b、 表明这个是维表，固定写法。
注意该表的来源是rds，后面的连接方式和正常的MYSQL连接方式没什么区别。
2.3、数据输出表的写法和维表的写法一致，只要提前在RDS上建好即可。
2.4、应用脚本开发：将引用到的业务表与维表进行关联，将数据输出至目标表
Insert into ads_product_qty
select c.dts_clomun3,
sum(b.dts_qty) as qty_sum
from table_a a
join ( select
dts_clomun1 ,
dts_skuname ,
dts_skucode ,
case
when dts_operation_flag = 'U' and dts_before_flag = 'Y' and dts_after_flag = 'N' then -1 * dts_qty
when dts_operation_flag = 'U' and dts_before_flag = 'N' and dts_after_flag = 'Y' then dts_qty
when dts_operation_flag = 'D' then -1 * dts_qty
else dts_qty
end as dts_qty
from table_b ) b
on a.dts_clomun1 = b.dts_clomun1
join div_project FOR SYSTEM_TIME AS OF PROCTIME() as c on b.dts_skucode = c.skucode -- a
group c.dts_clomun3
注解说明：
a、和标准SQL没太大区别，主要就是维表的使用方式略有不同，不过也是固定写法，照抄就行。
注意：由于原始数据有插入、删除、更新三种动作，所以DataHub上也会有三种状态的数据，这就需要分别进行处理，否则数据会不准。
3、 DataV开发：此处省略，简单归纳即：托一个图形，写一个SQL。

四、项目预案
由于流计算可能存在的风险，我们考虑以传统的计算方式开发第二套方案，当流计算出故障时，能快速切换方案，保证数据基本能正常使用，可能延迟会大一些。
通过评估，由于数据量预计不会太大，考虑使用定时调存储过程的方式计算指标到第二套输出表上，再开发第二套报表，展现和第一套一致。设计图如下：

图：定期调任务刷数据
通过实际测试，大部份指标能在半分钟内甚至十秒内出来，所以这些延迟勉强可以接受。由于技术实现并不复杂，此处略过。
五、项目压测
为了保障数据爆发时，平台依然能稳定的工作，我们进行了相关的一些测试。首先模拟大量的数据产生，确认数据从本地库至RDS的同步时间，RDS至DATAHUB的时间，以及流计算所处理的时间。
通过多次的测验得出：当数据量持续比较大的情况下，数据会有一些延迟，性能瓶颈主要体现在RDS以及DTS同步至DATAHUB环节，其中后一步比较明显。而平台稳定性基本上是可以的，流计算的处理效率非常令人满意。

最后更新：2017-11-13 16:34:45

  上一篇：  地平线余凯：自动驾驶处理器的“三国时代”| 清华人工智能研习社

  下一篇：  不忘初码，聚栈前行