blink cep基于用户的行为分析大杀器

本文涉及的产品
表格存储 Tablestore,50G 2个月
简介: #场景 今天我们做o2o有很多的线下场景需要基于用户的行为进行分析,比如 我们在商场入门口装了一个摄像头,可以通过摄像头识别商场进入的人脸,和出去时候的人脸,这样形成了2条数据。商场想统计下,每个用户在商场的逗留时间。这里面就出现了一种pattern的模式,就是当用户进门和出门这两个事件都发生的时候,激发某个动作(事件)。比如在这里是在用户出门的时候将用户进门时间和出门时间

场景

今天我们做o2o有很多的线下场景需要基于用户的行为进行分析,比如 我们在商场入门口装了一个摄像头,可以通过摄像头识别商场进入的人脸,和出去时候的人脸,这样形成了2条数据。商场想统计下,每个用户在商场的逗留时间。这里面就出现了一种pattern的模式,就是当用户进门和出门这两个事件都发生的时候,激发某个动作(事件)。比如在这里是在用户出门的时候将用户进门时间和出门时间的差值相减,并存储在tablestore当中。然后可以通过分析汇总算出今日用户在商场的平均逗留时间,继而可以统计出当月,当年的平均逗留时间,等等,促使商家提升总体运营水平。
这里面 用户进入商场 和用户走出商场是两个用户行为,当它们组合在一起时候就有了奇妙的意义。blink在分析此类用户行为方面提供了极其强大的模式匹配功能。下面我以这个场景,详细描述下如果使用。

架构图

image.png

用户行为表
image.png

用户逗留时间表
image.png

用户行为表用来记录摄像头识别的用户进门和出门的产生的记录,用户逗留时间表用来记录每个用户的逗留时间。
以用户行为表为blink的源表,用户逗留时间表为结果表。这里使用tablestore为存储数据库。为什么使用tablestore,因为tablestore的通道和blink能进行无缝的集成,tablestore的存储成本非常的低廉,扩展性又非常的好。有想了解tablestore的同学,点击传送门,当然也可以换成hbase等类似的数据库。
下面是具体的代码
CREATE TABLE mofun_source_user_action (
--id
id BIGINT,
-- 记录日期
record_date BIGINT,
--用户id
user_id BIGINT,
--数据触发时间
prd_time BIGINT,
--数据行为类型
opt_type varchar,
--计算列
ts AS to_timestamp(prd_time),
WATERMARK FOR ts AS withOffset(ts, 1000),
primary key (record_date, user_id, prd_time,opt_type)
)

CREATE TABLE ots_result_user_play_time (
--进入id
inid BIGINT,
--出去id
outid BIGINT,
--记录日期
record_date BIGINT,
user_id BIGINT,
play_time BIGINT,
primary key (record_date,user_id)
)

insert into ots_result_user_play_time
SELECT

inid,
outid,
start_tstamp,
`user_id`,
play_time

FROM mofun_source_user_action MATCH_RECOGNIZE (

 PARTITION BY `user_id`
 ORDER BY ts
 MEASURES  
            e1.id as inid,
            e2.id as outid,
            e1.record_date AS start_tstamp,
           (e2.prd_time-e1.prd_time) AS play_time
 ONE ROW PER MATCH
 PATTERN (e1->e2) WITHIN INTERVAL '10' Hour
 DEFINE
    e1 AS  e1.opt_type ='in',e2 as e2.opt_type='out'
 ) 

其中最关键的是最后这块代码
解释一下代码
DEFINE,定义的是模式匹配的变量,意思你要用哪列,什么条件的数据来进行条件匹配
PATTERN 就是变量的匹配模式,e1->e2的意思是指 进门后面如果出现了出门就匹配成功
MEASURES 里面是要在select中显示的数据列

具体文档看传送门(https://yuque.antfin-inc.com/rtcompute/doc/sql-query-cep),不过这篇文章讲得也不是很清楚,后面我会写一篇文章专门详细介绍复杂的匹配。

像上面这种场景稍微的进行一下更改就有很多场景可以使用。比如 我们现在经常使用短视频,那么我们怎么分析用户在某个短视频的停留时间呢?就可以在用户进入视频和出了视频产生一个事件,然后用上面这个语法就能分析出来,每个用户在视频上面的停留时间,然后根据排个实时排行榜,然后进行推荐。排行榜的架构,可以看我的上一篇文章,如何构建实时的排行数据。

相关实践学习
消息队列+Serverless+Tablestore:实现高弹性的电商订单系统
基于消息队列以及函数计算,快速部署一个高弹性的商品订单系统,能够应对抢购场景下的高并发情况。
阿里云表格存储使用教程
表格存储(Table Store)是构建在阿里云飞天分布式系统之上的分布式NoSQL数据存储服务,根据99.99%的高可用以及11个9的数据可靠性的标准设计。表格存储通过数据分片和负载均衡技术,实现数据规模与访问并发上的无缝扩展,提供海量结构化数据的存储和实时访问。 产品详情:https://www.aliyun.com/product/ots
相关文章
|
2月前
|
SQL 消息中间件 分布式计算
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践:恶意登录检测实现
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践:恶意登录检测实现
48 0
|
4月前
|
存储 SQL 算法
B端算法实践问题之Blink在实时业务场景下的优势如何解决
B端算法实践问题之Blink在实时业务场景下的优势如何解决
47 1
|
4月前
|
监控 Java API
【揭秘】如何用Flink CEP揪出那些偷偷摸摸连续登录失败的“捣蛋鬼”?——一场数据流中的侦探游戏
【8月更文挑战第26天】Flink 是一款先进的流处理框架,提供复杂事件处理(CEP)功能以识别实时数据流中的特定模式。CEP 在 Flink 中通过 `CEP` API 实现,支持基于模式匹配的事件检测。本文通过监测用户连续三次登录失败的具体案例介绍 Flink CEP 的工作原理与应用方法。首先创建 Flink 环境并定义数据源,接着利用 CEP 定义连续三次失败登录的模式,最后处理匹配结果并输出警报。Flink CEP 能够轻松扩展至更复杂的场景,如异常行为检测和交易欺诈检测等,有效应对多样化的业务需求。
47 0
|
5月前
|
数据采集 资源调度 搜索推荐
Flink在实时搜索引擎索引构建中的深度应用与实践
随着数据源规模的扩大和查询请求的增加,如何优化Flink的性能和资源调度成为了一个重要的问题。Flink提供了多种性能优化手段,如并行度调整、状态后端选择、任务链优化等。同时,Flink还支持与YARN、Kubernetes等集群管理系统集成,实现资源的动态调度和弹性伸缩,以适应不同规模的业务需求。
|
5月前
|
SQL 关系型数据库 MySQL
实时计算 Flink版产品使用问题之在Flink算子内部使用异步IO可以通过什么办法实现
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
6月前
|
SQL 关系型数据库 MySQL
实时计算 Flink版产品使用问题之自动调优指的是什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
7月前
|
SQL 数据处理 API
实时计算 Flink版产品使用合集之流处理过程中,对于某一条数据的异常,该如何处理
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
7月前
|
消息中间件 SQL Kafka
实时计算 Flink版产品使用合集之构建实时数据仓库时,如何操作在几分钟内一直变化的表
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
SQL 消息中间件 JSON
Flink CEP 新特性进展与在实时风控场景的落地
本次分享将会介绍 Flink 社区在 1.16 中对 Flink CEP 所做的增强与优化。
Flink CEP 新特性进展与在实时风控场景的落地
|
消息中间件 存储 分布式计算
基于 Apache Hudi 构建增量和无限回放事件流的 OLAP 平台
我们将讨论在构建流数据平台时如何利用 Hudi 的两个最令人难以置信的能力。 增量消费--每 30 分钟处理一次数据,并在我们的组织内构建每小时级别的OLAP平台 事件流的无限回放--利用 Hudi 的提交时间线在超级便宜的云对象存储(如 AWS S3)中存储 10 天的事件流(想象一个具有 10 天保留期的 kafka 主题) 具有部分记录更新的自定义 Hudi Payload 类
288 0
基于 Apache Hudi 构建增量和无限回放事件流的 OLAP 平台