作者：何孟强

1企鹅简介

杭州企鹅科技是一家集自主研发、商业运营、合作推广为一体的智慧物联网平台服务商。企鹅科技利用智能物联网技术平台为自助商业设备赋能，为运营商提供综合物联网转型解决方案和SaaS移动支付解决方案。公司依托领先的物联网技术，瞄准线下消费升级和差异化趋势，利用智能物联网技术连接赋能线下商业设备，实现商户智能化管理，降低运维成本，增加收入；同时为用户提供基于LBS定位的物联网自助智能体验服务，满足用户体验的智能化、个性化需求。

2. 业务场景

企鹅科技目前拥有约40万台物联网终端设备，包括洗衣机、烘干机、饮水机、花洒、吹风机、充电桩、按摩椅等。需要实时掌握终端设备的状态，发布指令后反馈结果，每天收发近4亿条消息。

以洗衣机场景为例：用户下单后，通过阿里巴巴云物联网平台向终端设备发送启动指令。指令发送成功后，需要实时了解洗衣机是否启动成功。以饮水机场景为例：用户扫描饮水机上的二维码并发出解锁命令后，需要实时了解饮水机的流量和状态。

在业务痛点：,由于终端设备数量过多，设备类型复杂，消息量巨大，实时性要求高，高频交易场景小，需要根据各种业务场景进行处理，IOT平台原有的通过直接消息订阅消息和消息流，无法满足我们复杂业务场景的高频小需求。

3.解决方案

为了解决上述业务痛点问题，选择阿里巴巴云实时计算Flink作为实时计算的基础设施，进行数据清洗、过滤和业务处理，然后输出到各个下游。

-h-arrow-right">系统架构：

数据流向及业务处理：

1. 设备信息数据实时上报到IOT物联网平台。

2. IOT物联网平台把数据流转到dataHUB。

3. 实时计算Flink版订阅DATAHUB数据。

4. 实时计算Flink根据业务规则进行数据清洗、加工输出至各下游端。

饮水场景解决方案：

饮水场景是一种小额的高频交易的后付费场景，用户下单后先下发解锁指令，让用户先使用然后再根据设备上报的流量进行计费。终端设备上报的信息中会有单笔订单上报，也会由于网络情况不好而挤压在终端的信息需同时进行上报，通过实时计算Flink可快速通过SQL方式进行数据过滤如：根据上报的类型和tag进行条件过滤如下：`type` = '08' AND `profile` IS NULL AND json_value ( `data`, '$.Flag' ) IN ( '04', '05' ); 可以判断该笔消息是饮水场景单笔订单的上报，很便捷的判断出消息的种类后输出至MQ、Redis等供下游消费使用，完成业务的闭环。

充电桩场景解决方案：

充电桩设备需实时掌握充电桩的通道的功率以及充电桩的状态及充电的剩余时间，为实时掌握通道的功率我们通过实时计算Flink进行加工后数据输出至TSDB中。通过实时计算Flink可快速完成业务需求。

定义datahub的源表信息和TSDB的结果表信息

--创建iot_up数据源表create table iot_up ( `id` varchar, `imei` varchar, `pk` varchar, `type` varchar, `data` varchar, `profile` varchar) with ( type = 'datahub', endPoint = 'http://dh-cn-shanghai.aliyun-inc.com', project = '***', topic = '***', accessId = '***', accessKey = '***');--创建tsdb表CREATE TABLE charge_Power ( `metric` VARCHAR, `timestamp` INTEGER, `value` INTEGER, `imei` VARCHAR, `channel` VARCHAR, `msgid` VARCHAR, `pk` VARCHAR) WITH ( type='hitsdb', host='***', port='***');

根据业务规则过滤后输出至TSDB中

--创建tsdb表写入数据insert into charge_power select 'power', cast(UNIX_TIMESTAMP(`time`) as int), cast(json_value(`data`,'$.Powers[0].Power') as int), imei, json_value(`data`,'$.Powers[0].Channel'), json_value(`data`,'$.Powers[0].MsgId'), pk from iot_up where `type` = '0706' and json_value(`data`,'$.Powers[0]') is not null;