标签: 39期

一、《flink实时规则营销系统》项目背景传统的“精准营销平台”由营销人员基于画像标签数据去圈选人群，进行营销推送，存在不足；这种传统手段，往往无法抓住那些“转瞬即逝的营销机会”；如：一个促销狂热型用户，正在浏览包包时，及时推送出正在做促销活动的包包信息，是最有效的；如：一个价格敏感型用户，正在犹豫不决反复查看他购物车中的某目标商品时，适时推送优惠券，是最有效的；这些场景，显然无法通过营销人员通过人工手段去发现，需要用软件系统自动、实时监控用户的行为，并实时做出判断，并进而驱动营销消息推送系统推送相关营销内容；总之，在应对这一类的需求时，以前的传统的“基于用户画像标签库的精准营销平台”显得力不从心；因而，擅长用键鼠改变世界的攻城狮们，决定为公司的推广、运营部门开发一个应对上述场景的自动化、智能的、实时、规则动态可变的营销利器——《实时动态规则智能运营系统》二、需求分析2.1营销规则举例-搜索兴趣词包含“商务休闲”的用户，如果浏览包含“商务休闲”的商品页面，则触发一个优惠券推送消息；-活跃度等级为C的用户，且在规则发布前的3天内，发生过“搜索雀巢”AND“加购”，当他再次上线时，触发一个app弹窗消息；-活跃度等级为C的用户，规则发布前2天内，发生过A事件3次以上，规则上线后发生了B事件，且之后间隔1天再次发生B事件，则触发；-规则运行时，检测到用户发生了下单事件，而在随后的15分钟内没有支付，则触发；-规则运行时，一个用户如果搜索的关键词符合某正则表达式，则触发；-宝妈用户，在2022-08-07开始，做过A行为3+次，做过（B行为，E行为）序列2+次，则触发；2.2规则营销需求的宏观抽象-营销规则有多样化的灵活的定义（不光参数可以支持自定义，且运算模型也可以自定义）；-营销规则通常针对满足某画像的人群，也可没有针对性-营销规则通常都有有效期（也有可能是无限期的），要求系统能够支持动态增加、停用运营计划（规则）-营销规则根据需求的变化，需要能够方便地进行在线管理，如新增、停用、修改三、行为判断的逻辑分析1）行为判断：行为事件的类型及属性的判断判断用户的某个行为事件，是否是规则条件中要求的事件-简单的，可能就是 event_id="X"   ANDproperties["p1"]=/>/!=/<"v1"-稍微复杂的，可能就是  event_id="X"   AND  properties["p1"] 包含"kw1"-再复杂点的，可能就是  event_id="X"   AND properties["p1"] +properties["p2"]>100   ANDproperties["p3"]满足正则表达式 [1,10]*?a2）行为序列的判断判断用户的一系列行为事件中，是否满足规则条件所要求的依次发生的序列，比如A  B  E行为序列判断，是上述“行为判断”的衍生本质上，就是在判断事件的基础上，加上时间先后的序列关系这个逻辑，可以用正则匹配进行计算（全窗口运算）； 也可以用滚动聚合的方式计算；3）事件（序列）发生次数的判断判断要求的事件或事件序列，发生了多少次是上述“行为判断”、“行为序列判断”的衍生本质上，就是在判断事件（或序列）的基础上，进行次数累计四、画像条件判断的实现难点初略考虑，应该就是查询用户画像标签库即可；站在系统的角度来考虑时：什么时机去查询？ 去哪里查？比如： 查询时机： 当一个人的某个行为触发了某个规则，此时去查询该用户的画像标签是否满足规则的人群画像条件查询的目标： 用户画像标签数据库如果画像数据库在hbase中，可以用如下伪代码来快速判断##如果条件是：活跃等级=C  AND   首单时间<"2022-06-01"get=newGet(guid)get.addColumn("活跃等级")get.addColumn("首单时间")result=table.get(Get)activeLevel=result.getValue("活跃等级")firstOrderDate=result.getValue("首单时间")if(activeLevel='C' && firstOrderDate<"2022-06-01") returntrue;##如果条件是：兴趣词：包含"速溶咖啡"##这种条件，用hbase的支撑就不太好处理了//favroitWords:[海底捞,咖啡伴侣,小罐速溶咖啡，乔丹，篮球]favroitWords=result.getValue("兴趣词")  

flink实时规则营销系统（39期，2023版）——简单来说，本系统核心是一个基于事件驱动且可进行动态规则计算的实时系统，在技术上它是通用的；本套架构及系统内核，不仅可以用于“实时运营”，也可以用于“实时风控”，“实时推荐”，“实时交通监控”等场景。运营场景举例：下单促付款，新品发布信息精准送达。本系统，在核心规则引擎之外，也集成了大量metric及跟踪数据输出，能在系统中及系统外进行多维度数据分析，以深度掌握各类市场运营活动的效果，以及系统运行的各类状态什么是flink?flink是一个分布式，高性能，随时可用的以及准确的流处理计算框架，flink可以对无界数据（流处理）和有界数据（批处理）进行有状态计算（flink天生支持状态计算）的分布式，高性能的计算框架。flink的基石flink的四大基石：checkpoint,state,time,windowcheckpoint:基于chandy-lamport算法实现分布式计算任务的一致性语义；state:flink中的状态机制，flink天生支持state,state可以认为程序的中间计算结果或者是历史计算结果；time:flink中支持基于事件时间和处理时间进行计算，sparkstreaming只能按照processtime进行处理；基于事件时间的计算我们可以解决数据迟到和乱序等问题。window:flink提供了更多丰富的window,基于时间，基于数量，sessionwindow,同样支持滚动和滑动窗口的计算。Flink的应用Flink是一个大数据流处理引擎，它可以为不同的行业提供大数据实时处理的解决方案。随着Flink的快速发展完善，如今在世界范围许多公司都可以见到Flink的身影。目前在全球范围内，北美、欧洲和金砖国家均是Flink的应用热门区域。当然，这些地区其实也就是IT、互联网行业较发达的地区。Flink在国内热度尤其高，一方面是因为阿里的贡献和带头效应，另一方面也跟中国的应用场景密切相关。中国的人口规模与互联网使用普及程度，决定了对大数据处理的速度要求越来越高，也迫使中国的互联网企业去追逐更高的数据处理效率。试想在中国，一个网站可能要面对数亿的日活用户、每秒数亿次的计算峰值，这对很多国外的公司来说是无法想象的。而Flink恰好给我们高速准确的处理海量流式数据提供了可能。Flink环境部署:要了解一个系统，一般都是从架构开始。我们关心的问题是：系统部署成功后各个节点都启动了哪些服务，各个服务之间又是怎么交互和协调的。当Flink集群启动后，首先会启动一个JobManger和一个或多个的TaskManager。由Client提交任务给JobManager，JobManager再调度任务到各个TaskManager去执行，然后TaskManager将心跳和统计信息汇报给JobManager。TaskManager之间以流的形式进行数据的传输。上述三者均为独立的JVM进程。Client为提交Job的客户端，可以是运行在任何机器上（与JobManager环境连通即可）。提交Job后，Client可以结束进程（Streaming的任务），也可以不结束并等待结果返回。JobManager主要负责调度Job并协调Task做checkpoint，职责上很像Storm的Nimbus。从Client处接收到Job和JAR包等资源后，会生成优化后的执行计划，并以Task的单元调度到各个TaskManager去执行。TaskManager在启动的时候就设置好了槽位数（Slot），每个slot能启动一个Task，Task为线程。从JobManager处接收需要部署的Task，部署启动后，与自己的上游建立Netty连接，接收数据并处理。本地模式环境部署步骤如下：#1.下载安装包并上传到/usr/local/src目录#2.解压安装包并重命名为flink[root@nodesrc]$tar-zxfflink-1.14.4-bin-scala_2.12.tgz[root@nodesrc]$tar-zxfjdk-8u111-linux-x64.tar.gz#3.修改安装包所属用户和用户组权限[root@nodesrc]$chown-Rroot.rootflink-1.14.5#4.配置Flink环境变量并重新加载使其生效[root@nodesrc]$vim~/.bash_profileexportJAVA_HOME=/usr/local/src/jdk1.8.0_111/exportFLINK_HOME=/usr/local/src/flink-1.14.5/exportPATH=$PATH:$JAVA_HOME/bin:$FLINK_HOME/bin[root@nodesrc]$source~/.bash_profile#4.启动"集群"[root@nodesrc]$start-cluster.shStartingcluster.Startingstandalonesessiondaemononhostnode.Startingtaskexecutordaemononhostnode.[root@nodesrc]$jps17186Jps17078TaskManagerRunner16809StandaloneSessionClusterEntrypoint#5.访问WebUI界面8081端口查看运行情况分发Flink安装包配置到另外两个节点使用Linuxscp命令把node01节点的配置分发到另外两个节点上。#使用scp分发[root@node01conf]$cd/usr/local[root@node01local]$scp-rflinknode02:/usr/local/[root@node01local]$scp-rflinknode03:/usr/local/启动Flink集群#批量统一启动[root@node01bin]$start-cluster.shStartingcluster.Startingstandalonesessiondaemononhostnode01.Startingtaskexecutordaemononhostnode01.Startingtaskexecutordaemononhostnode02.Startingtaskexecutordaemononhostnode03. #单独启动[root@node01bin]$ jobmanager.shstart/stop[root@node01bin]$ taskmanager.shstart/stop #使用脚本查看服务进程[root@node01bin]$jps