【最佳实践】实时计算Flink在IoT行业的实时数仓建设实践

2021-09-17 22:30:11

行业背景

  • 行业现状： 

    • 物联网(Internet of Things，以下简写为 IoT)是互联网、传统电信网等资讯的承载体，让所有具备独立功能的普通物体实现互联互通的网络。物联网广泛应用于运输和物流、健康医疗、智慧环境(家庭、办公、工厂)等领域， 具有十分广阔的市场和应用前景。物联网将智能感知、识别技术、网络通信与普适计算等技术融合起来，被认为是继计算机、互联网、智能手机之后世界信息产业发展的下一个风口。
    • 据 IDC 估计，到 2020 年物联网将在全球范围内产生 1.46 万亿美元的价值。据预测，届时中国的物联网市场规模将超过 1.8 万亿人民币。得益于庞大的人口基数和低廉的芯片制造成本，中国将成为物联网行业的主要参与者，并在推动全球物联网市场增长上发挥重要的作用。

  • 大数据在其行业中的作用：

    • 数以百万计的物联网设备连接到物联网，产生了大量的数据，通过大规模分析这些数据了解影响业务的上下文关系和模式，从而做出更加实时决策，因此，可以说大数据和物联网密切相关。
    • 物联网大数据分析可以帮助人们更好地理解数据，从而做出更有效、明智的决定。大数据分析使数据挖掘者和科学家能够利用传统工具分析非结构化数据。此外，大数据分析的目的是利用数据挖掘技术，提取知识信息，这些信息有助于进行预测，识别趋势，发现隐藏的信息，并做出决定。

业务场景

某公司开发了一套针对商场的人流量管理和分析系统。当顾客进入商场后，固定部署的WIFI探针将实时探测并采集顾客手机或PC终端的WIFI信息。这些顾客WIFI信息经过Flink实时统计和分析后获得商场中客流量、客流量的高峰期、顾客喜欢哪些商店、新客与常客的比例、商场的成交额、顾客的实时位置等等指标。客流量指标可用于数据大屏，顾客的实时位置指标可用于展示顾客实时密度图，顾客喜欢的商店指标可用于个性化的广告推送应用等。

技术架构

【最佳实践】实时计算Flink在IoT行业的实时数仓建设实践
架构解析：
数据采集：该场景中，数仓的数据来源于WIFI探针探测到的顾客手机和PC终端的WIFI信息，实时采集至DataHub作为Flink的输入数据。
实时数仓架构：该场景中，整个实时数仓的ETL和BI部分的构建，全部通过Flink完成，Flink实时读取DataHub的数据进行处理，并与维表进行关联查询等操作，统计和分析结果最终输出到RDS和TableStore等存储系统，以供业务系统使用。

业务指标

  • 场景一：客流量统计

    • 每日商场客流量
    • 每日商店客流量
    • 顾客的实时位置
    • 每日商场新增顾客人数
    • 每日商场中顾客数量前5的商店
    • 每日商场中顾客最多的时间段及顾客数量

  • 场景二：顾客喜好分析

    • 单日顾客最喜欢的商店
    • 单日顾客进入超过一次的商店

业务代码

场景一：客流量统计

每日商场客流量

对顾客WIFI信息按天维度进行分组，使用count distinct语法去重统计得到商场中顾客人数。

输入表

--顾客WIFI信息
CREATE TABLE user_wifidata (
    id                            varchar,
    shop_code                     varchar, --商店编码
    ap_ip                         varchar, --WIFI的ip地址
    occur_time                    varchar, --记录时间
    user_device_mac               varchar, --终端mac地址
    rssi                          varchar, --接收信号强度
    ap_mac                        varchar, --WIFI的MAC地址
    pt                            varchar, --pt
    ts_occur_time AS TO_TIMESTAMP(cast(cast(occur_time as bigint)*1000 as bigint)),
    WATERMARK FOR ts_occur_time AS withOffset(ts_occur_time, 60000)
) WITH (
  type = 'datahub',...
);

输出表

--每日商场客流量
CREATE TABLE trading_area_daily_people (
    count_date                      varchar, --日期
  people_sum                                            bigint,     --客流量
    primary key (count_date)
) WITH (
    type='rds',...
);

业务代码

--每日商场可流量统计
INSERT into trading_area_daily_people
select 
    DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd') count_date,
    count(distinct data.user_device_mac)
from
user_wifidata data
group by DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd')
;

每日商店的客流量

顾客WIFI信息通过关联商场中的商店维表获得商店名称，在此基础上按照商店名称和按天维度时间字段进行分组，使用count distinct语法去重统计每日商店的顾客人数。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--每日商店的客流量
CREATE TABLE shop_area_daily_people (
    count_date                      varchar,  --日期
  shop_name                                                varchar,    --商店名
  people_sum                                            bigint,      --客流量
    primary key (count_date)
) WITH (
    type='rds',...
);

维表

--商店维表
CREATE TABLE area_shop (
    id                                      bigint,
    shop_code             varchar, --商店编码
  shop_name                      varchar, --商店名
    PRIMARY KEY (shop_code),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);

业务代码

--每日商店的客流量统计
INSERT into shop_area_daily_people
select 
    DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd') count_date,
      mas.shop_name,
    count(distinct data.user_device_mac)
from
user_wifidata data
left outer join area_shop FOR SYSTEM_TIME AS OF PROCTIME() mas 
on data.shop_code=mas.shop_code 
group by DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd'),mas.shop_name
;

顾客的实时位置

顾客WIFI信息通过关联探测顾客位置维表（place_info），获取到顾客的实时位置信息，通过1秒钟的滚动窗口实现每秒钟上报一次位置信息，实时投放到数据大屏上，从而实现顾客实时密度图功能。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--顾客来店数据表
CREATE TABLE area_customer_info_allday (
    user_device_mac        varchar, --终端mac地址
    come_time              varchar, --到来的时间
  ap_mac                               varchar, --WIFI的MAC地址
    occur_time                       varchar, --记录时间
    used_x                                bigint,  --顾客的经度
    used_y                                bigint,  --顾客的维度
    first_come_flag                 varchar, --是否是第一次
    rssi                               varchar, --接收信号强度
    save_time                         varchar, --时间
    pt                                   varchar, --pt
  PRIMARY KEY (user_device_mac)
) WITH (
  type='ots'
);

维表

--探测设备可以获取到顾客的位置
CREATE TABLE place_info (
    user_device_mac               varchar,  --终端mac地址
    used_x                        bigint,        --顾客的经度
    used_y                        bigint,        --顾客的维度
    min_rssi                      bigint,     --最小的接收信号强度
    PRIMARY KEY (user_device_mac),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);

业务代码

--通过1秒钟的滚动窗口实现每秒钟上报一次位置信息
CREATE view record_sec
(
  user_device_mac          ,
    come_time                ,
    ap_mac                             ,
    occur_time                    ,
    user_loca_x                    ,
    user_loca_y                    ,
    first_come_flag            ,
    rssi                                ,
    save_time                        ,
    shop_code                        ,
    pt
) as 
select data.user_device_mac ,
DATE_FORMAT(TUMBLE_START(data.ts_occur_time, INTERVAL '1' SECOND),'yyyy-MM-dd HH:mm:ss') ,
data.ap_mac,
data.occur_time,
'',
'',
'',
min(data.rssi),
FROM_UNIXTIME(unix_timestamp()),
grd.shop_code,
max(data.pt) 
from  
user_wifidata data 
GROUP BY TUMBLE(ts_occur_time, INTERVAL '1' SECOND),ap_mac,user_device_mac,shop_code,occur_time
;

CREATE view view1 as
select
a0.user_device_mac,
a0.come_time,
a0.ap_mac,
a0.occur_time,
mai.used_x ,
mai.used_y ,
a0.first_come_flag,
a0.rssi,
a0.save_time,
a0.pt
from  
record_sec a0 
left OUTER  join place_info FOR SYSTEM_TIME AS OF PROCTIME() mai 
on mai.user_device_mac=a0.user_device_mac;

INSERT into area_customer_info_allday
select *                    
 from view1 ;

每日商场新增顾客人数

顾客WIFI信息通过关联探测顾客位置维表（place_info），获取到顾客的实时位置信息，然后通过关联顾客来店记录维表，获取到顾客是否是新顾客信息，统计新顾客的人数。同时将新顾客信息更新到顾客来店记录表中标注成常客。此场景下，顾客来店记录表在同一个作业中即作为结果表又作为维表。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--每日商场新增顾客人数
CREATE TABLE customer_info_people_sum (
  people_sum                       bigint,      --新增顾客人数
  come_time                            varchar,    --日期
PRIMARY KEY (come_time)
) WITH (
    type='rds',...
);

--顾客来店记录表
CREATE TABLE shop_user_record_update (
    user_device_mac     varchar,          --终端mac地址
  first_come_flag          varchar ,            --是否是第一次来
  first_coming_time     varchar,          --进店时间
  last_coming_time      varchar ,            --离开时间
    PRIMARY KEY (user_device_mac),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='ots',
  tableName='shop_user_record'...
);

维表

--使用场景一“商店中顾客的实时位置”的维表：place_info
CREATE TABLE place_info (
    user_device_mac               varchar, --终端mac地址
    used_x                        bigint,     --顾客的经度
    used_y                        bigint,  --顾客的维度
    min_rssi                      bigint,  --最小的接收信号强度
    PRIMARY KEY (user_device_mac),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);  

--顾客来店记录
CREATE TABLE shop_user_record (
    user_device_mac     varchar,  --终端mac地址
  first_come_flag     varchar,    --是否是第一次来
  first_coming_time     varchar,  --进店时间
    last_coming_time       varchar,    --离开时间
    PRIMARY KEY (user_device_mac),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='ots',
  tableName='shop_user_record'...
);

业务代码

--通过1秒钟的滚动窗口实现每秒钟上报一次位置信息
CREATE view record_sec
(
  user_device_mac,
    come_time             ,
    ap_mac                  ,
    occur_time         ,
    user_loca_x         ,
    user_loca_y         ,
    first_come_flag,
    rssi                     ,
    save_time             ,
    shop_code             ,
    pt
) as 
select data.user_device_mac ,
DATE_FORMAT(TUMBLE_START(data.ts_occur_time, INTERVAL '1' SECOND),'yyyy-MM-dd HH:mm:ss') ,
data.ap_mac,
data.occur_time,
'',
'',
'',
min(data.rssi),
FROM_UNIXTIME(unix_timestamp()),
grd.shop_code,
max(data.pt) 
from  
user_wifidata data
GROUP BY TUMBLE(ts_occur_time, INTERVAL '1' SECOND),ap_mac,user_device_mac,shop_code,occur_time
;

CREATE view view1 as
select
a0.user_device_mac,
a0.come_time,
a0.ap_mac,
a0.occur_time,
mai.used_x ,
mai.used_y ,
a0.first_come_flag,
a0.rssi,
a0.save_time,
a0.pt
from  
record_sec a0 
left OUTER  join place_infomst_buynoplace_infow_ap_info FOR SYSTEM_TIME AS OF PROCTIME() mai 
on mai.user_device_mac=a0.user_device_mac;

CREATE view view2 as
select
a1.user_device_mac,
a1.come_time,
a1.ap_mac,
a1.occur_time,
cast(a1.used_x as varchar) as used_x,
cast(a1.used_y as varchar) as used_y,
mci.first_come_flag as first_come_flag,
a1.rssi,
a1.save_time,
a1.pt,
mci.first_coming_time,
mci.last_coming_time
from  
view1 a1
left OUTER  join shop_user_record FOR SYSTEM_TIME AS OF PROCTIME() mci 
on mci.user_device_mac = a1.user_device_mac ;


CREATE view record_out as
select
a2.user_device_mac as user_device_mac,
a2.come_time  as come_time,
a2.occur_time as occur_time,
a2.ap_mac as ap_mac,
a2.used_x as used_x,
a2.used_y as used_y,
if(a2.first_come_flag is null ,'1','0') as first_come_flag,--first_come_flag,
a2.rssi as rssi,
a2.save_time as save_time,
a2.pt as pt,
a2.first_coming_time as first_coming_time,
a2.last_coming_time as last_coming_time
from 
view2 a2 ;

--更新shop_user_record维表的顾客来店记录
INSERT into shop_user_record_update  
select user_device_mac , first_come_flag,first_coming_time,  last_coming_time           
 from record_out ;

--统计每日商店新顾客数
insert into customer_info_people_sum
select t1.people_sum,t1.come_time
from (select  count(distinct dt1.user_device_mac) as people_sum, DATE_FORMAT(dt1.come_time,'yyyy-MM-dd') as come_time
            from record_out dt1 where  dt1.first_come_flag='1' 
      group by DATE_FORMAT(dt1.come_time,'yyyy-MM-dd')) t1

每日商场中顾客数量前5的商店

顾客WIFI信息按照天维度和商店分组后，统计每日商店顾客数量，通过topn语句获取到每日顾客数量前5的商店信息，再通过关联商场店铺表得到商店名。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--每日商场中顾客数量前5的商店
CREATE TABLE top5_shop (
        shop_code   varchar, --商店编码
    shop_name     varchar, --商店名
    date_time     varchar, --记录顾客进入商店的时间（天维度）
    people_sum  bigint,  --顾客数量
    PRIMARY KEY (shop_code)
) WITH (
  type='rds'...
);

维表

--商场店铺表：使用场景一“每日商店的客流量”的维表：area_shop
CREATE TABLE area_shop (
    id                              bigint,
    shop_code         varchar, --商店编码
  shop_name                  varchar, --商店名
    PRIMARY KEY (shop_code),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);

业务代码

--按天、商店分组统计商店的顾客数
CREATE VIEW Window1 AS
SELECT shop_code,
       DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd') AS date_time,
       count(distinct user_device_mac) AS people_sum
FROM user_wifidata data
GROUP BY shop_code, DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd');

-- 统计每天top5客流量的商店
CREATE VIEW top5_view AS
SELECT shop_code, date_time, people_sum, rownum FROM
(
   SELECT
   shop_code, date_time, people_sum,
   ROW_NUMBER() OVER (PARTITION BY shop_code,date_time ORDER BY people_sum DESC) as rownum
   FROM
   Window1
)
WHERE rownum <= 5;

INSERT into top5_shop
select 
    top5.shop_code ,
    mas.shop_name ,
    top5.date_time ,
    top5.people_sum
from
top5_view top5
inner join area_shop FOR SYSTEM_TIME AS OF PROCTIME() mas 
on top5.shop_code=mas.shop_code

每日商场中顾客最多的时间段及顾客数量

顾客WIFI信息按照1小时滚动窗口统计商场顾客数量，使用topn语句获取到每日商场中顾客最多的时间段。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--商场中顾客最多的时间段及顾客数量
CREATE TABLE top1_time (
    start_time timestamp, --时间段
    people_sum bigint ,   --顾客数量
    PRIMARY KEY (start_time)
) WITH (
  type='rds',...
);

业务代码

--1小时滚动窗口统计每小时商场的顾客数
CREATE VIEW Window1 AS
SELECT shop_code,
      TUMBLE_START(ts_occur_time, INTERVAL '1' hour) AS start_time,
      count(distinct user_device_mac) AS people_sum
FROM user_wifidata
GROUP BY shop_code, TUMBLE(ts_occur_time, INTERVAL '1' hour);

INSERT into top1_time 
SELECT start_time, people_sum FROM
(
   SELECT
   start_time, people_sum,
   ROW_NUMBER() OVER (PARTITION BY start_time ORDER BY people_sum DESC) as rownum
   FROM
   Window1
)
WHERE rownum <= 1;

场景二：顾客喜好分析

每日单个顾客最喜欢的商店

顾客WIFI信息按照商店、顾客、天维度进行分组得到每日顾客进入商店的次数，使用topn语句获取到每日顾客去过最多的商店，通过关联商店名维表获取商店名，从而统计每日单个顾客最喜欢的商店。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--单个顾客最喜欢的商店输出表
CREATE TABLE favorite_shop (
      user_device_mac varchar,        --终端mac地址
    date_time             varchar,         --记录顾客进入商店的时间（天维度）
    shop_code             varchar,       --商店编码
    shop_name             varchar,        --商店名
    shop_number       bigint ,        --进店次数
PRIMARY KEY (shop_code)
) WITH (
  type='rds'...
);

维表

--商场店铺表：使用场景一“每日商店的客流量”的维表：area_shop
CREATE TABLE area_shop (
    id                              bigint,
    shop_code         varchar, --商店编码
  shop_name                  varchar, --商店名
    PRIMARY KEY (shop_code),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);

业务代码

--按天、顾客和商店分组统计单个客户当天进入单个商店的次数
CREATE VIEW Window1 AS
SELECT user_device_mac,
            shop_code,
      DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd') AS date_time,
      count(shop_code) AS shop_number
FROM user_wifidata data
GROUP BY user_device_mac,shop_code, DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd');

-- 统计一天中顾客进入次数最多的商店，并输出
CREATE VIEW top1_view AS
SELECT user_device_mac,shop_code, date_time, shop_number, rownum FROM
(
   SELECT
   user_device_mac,shop_code, date_time, shop_number,
   ROW_NUMBER() OVER (PARTITION BY user_device_mac,shop_code,date_time ORDER BY shop_number DESC) as rownum
   FROM
   Window1
)
WHERE rownum <= 1;


--关联商店名维表并输出
INSERT into favorite_shop
select 
        top1.user_device_mac,
    top1.date_time ,
    top1.shop_code ,
    mas.shop_name ,
    top1.shop_number
from
top1_view top1
inner join area_shop FOR SYSTEM_TIME AS OF PROCTIME() mas 
on top1.shop_code=mas.shop_code

单日单个顾客进入超过一次的商店

顾客WIFI信息按照商店、顾客、天维度进行分组获取到每日顾客进入商店的次数，通过where语句获取到每日顾客进入超过一次的商店，再通过关联商店名维表得到商店名，从而得到单日单个顾客进入超过一次的商店信息。

输入表

--顾客WIFI信息：使用场景一“每日商场客流量”的输入表：user_wifidata

输出表

--单日单个顾客进入超过一次的商店
CREATE TABLE morethanonce_shop (
      user_device_mac varchar,                --终端mac地址
    date_time             varchar,                 --记录顾客进入商店的时间（天维度）
    shop_code             varchar,               --商店编码
    shop_name             varchar,                --商店名
    shop_number       bigint,                 --进店次数
  PRIMARY KEY (user_device_mac)
) WITH (
  type='rds',...
);

维表

--商场店铺表：使用场景一“每日商店的客流量”的维表：area_shop
CREATE TABLE area_shop (
    id                              bigint,
    shop_code         varchar, --商店编码
  shop_name                  varchar, --商店名
    PRIMARY KEY (shop_code),
    PERIOD FOR SYSTEM_TIME
) WITH (
    type='rds',...
);

业务代码

--按天、顾客和商店分组统计单个客户当天进入单个商店的次数
CREATE VIEW Window1 AS
SELECT user_device_mac,
            shop_code,
      DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd')AS date_time,
      count(shop_code) AS shop_number
FROM user_wifidata data
GROUP BY user_device_mac,shop_code, DATE_FORMAT(data.ts_occur_time,'yyyy-MM-dd');

-- 1天中顾客进入超过两次的商店
CREATE VIEW exceed2_view AS
SELECT user_device_mac,shop_code, date_time, shop_number FROM
Window1
WHERE shop_number >= 2;

INSERT into morethanonce_shop
select 
        exceed2.user_device_mac,
    exceed2.date_time ,
    exceed2.shop_code  ,
    mas.shop_name          ,
    exceed2.shop_number
from
exceed2_view exceed2
inner join area_shop FOR SYSTEM_TIME AS OF PROCTIME() mas 
on exceed2.shop_code=mas.shop_code

实时计算 Flink 版产品交流群

【最佳实践】实时计算Flink在IoT行业的实时数仓建设实践

阿里云实时计算Flink - 解决方案：
https://developer.aliyun.com/article/765097
阿里云实时计算Flink - 场景案例：
https://ververica.cn/corporate-practice
阿里云实时计算Flink - 产品详情页：
https://www.aliyun.com/product/bigdata/product/sc

上一篇：《T-SQL性能调优秘笈——基于SQL Server 2012 窗口函数》——1.6　窗口定义的重复使用


下一篇：Hologres揭秘：如何支持超高QPS在线服务（点查）场景