标签: Flink

一、《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

什么是风控系统？风控系统是指一系列的完整的风险控制，以保证事情向好的方向发展，而免受不可预估的经济和财产损失而措手不及。风控类型风控类型一般分为事前、事中、事后三种。事前风控是指在交易指令发送到交易所前，对交易指令进行风险检测，通过检测的交易指令则提交到交易模块进行报单，未通过检测的交易指令将直接予以拒绝。对于追求低延时的交易策略，事前风控需要在极短的时间内完成。事中风控主要是指在交易过程中，交易团队对策略的信号生成、执行情况进行监控以及盘中对策略的风险度进行实时监控。事后风控是对交易数据在盘后进行分析，比如策略算法是否存在错误、策略的回撤是否可控、是否有计划外的持仓出现等，从而制定更严谨的业务风控预案和优化代码算法调整策略表现。在src文件夹中创建一个新文件，文件命名为“AppState.tsx”，我们将会在这个文件中创建全局state和上下文组件contexcomponent。首先，引入我们需要的react框架。interfaceAppStateValue{username:string,shoppingCart:{items:{id:number;name:string}[];};}然后，打开Header.tsx，在这个文件中我们创建一个新的函数式react组件，顺便引入我们的css样式文件importReactfrom"react";importstylesfrom"./Header.module.css";exportconstHeader:React.FC=()=>{return()}那么接下来，我们来把signin页面从路径JSX代码中挪出去，以页面组件的形式渲染出来。既然是页面，那么我们在pages文件夹中创建signin的相关页面吧。在pages文件夹中创建子文件夹signin，然后创建我们的三大金刚，index.ts,SignInPage.tsx，以及SignInPage.module.css。请同学们先打开SignInPage.tsx，我们先为页面创建一个最简单的函数式组件，组件内渲染一个H1标题，标题名称登陆页面。importReactfrom"react";exportconstSignInPage:React.FC=()=>{  return<h1>SignIn</h1>;};打开DetailPage.tsx，我们先引入react框架，创建一个最基本的函数式组件importReactfrom"react";exportconstDetailPage:React.FC=(props)=>{return(<div><h1>旅游路线详情页面</h1></div>);};这个get的返回值是一个promise。我们需要使用.then函数来处理，函数的参数是个lamda表达式，而我们关心的是响应的主体数据，也就是reponse.data，可以简单使用花括号在参数中直接展开reponse，取得data。理论上来说，响应数据data的类型应该与postman中的输出结果是一摸一样的，所以，我们把数据添加到组件state的产品列表1、2、3中。componentDidMount(){axios.get('http://123.56.149.216:8080/api/productCollections',{headers:{'x-icode':'FB80558A73FA658E',},}).then(({data})=>{this.setState({productList1:data});})}函数体内使用async，这个函数本身也必须是async，所以我们给componentDidMount加上async关键词。因为axios.get的返回值本来就是一个promise，所以我们可以给它加上await，等待get请求的执行，并取得响应数据response，而我们需要的是响应的主体数据data，可以使用花括号直接展开。于是，我们就获得了与postman一样的数据，接下来的代码就和promisethen一样了，稍微整理一下。asynccomponentDidMount(){const{data}=awaitaxios.get("http://123.56.149.216:8080/api/productCollections",{headers:{"x-icode":"FB80558A73FA658E",},});this.setState({loading:false,productList:data});}但是如果api访问错误。我们就需要使用trycatch来捕获error了，出错的时候我们同样要吧loading设置为false，但是这个时候error就要被切换为错误信息了，e.message.interfaceStateProps{loading:boolean,productList:any[],}constructor(props){super(props)this.state={loading:true,error:null,productList:[],};}asynccomponentDidMount(){try{const{data}=awaitaxios.get("http://123.56.149.216:8080/api/productCollections");this.setState({loading:false,error:null,productList1:data,});}catch(e){this.setState({loading:false,error:e.message,});}}Ok,转菊花的控制状态配置完成，接下来就要在jsx代码中转菊花了，从组件state中展开loading和error。我们依旧使用最简单的方法，直接暴力使用一个if语句，当loading为ture的时候输出转菊花，就是渲染Spin组件，设置size大小为最大，“large”。然后定义一些inlinestyleif(loading){return(<Spinsize="large"style={{marginTop:200,marginBottom:200,marginLeft:"auto",marginRight:"auto",width:"100%",}}/>);}接下来，我们来复制粘题一下代码。请同学们打开DetailPage。我们把useEffect中所有的数据请求代码全部剪切出来。复制到getProductDetail的callback函数中去。接着给dispatch加上thunkAPI。然后还有一个报错，我们需要引用axiosimport axios from "axios";  export const getProductDetail=createAsyncThunk("productDetail/getProductDetail",async(touristRouteId:string,thunkAPI)=>{thunkAPI.dispatch(productDetailSlice.actions.fetchStart());try{const{data}=await axios.get(`http://123.56.149.216:8080/api/touristRoutes/${touristRouteId}`);thunkAPI.dispatch(productDetailSlice.actions.fetchSuccess(data));}catch(error){thunkAPI.dispatch(productDetailSlice.actions.fetchFail(error.message));}});在callback函数中去掉所有的dispatch，然后也删掉trycatch，因为我们将会返回promise，让promise自己来处理。所以retun的应该是data这个对象。export const getProductDetail=createAsyncThunk("productDetail/getProductDetail",async(touristRouteId:string,thunkAPI)=>{const{data}=await axios.get(`http://123.56.149.216:8080/api/touristRoutes/${touristRouteId}`);return data;});

Flink从0到1实战实时风控系统课程下载2023需要解决的问题哪些是风险事件，注册、登录、交易、活动等事件，需要业务埋点配合提供实时数据接入什么样的事件是有风险的，风险分析需要用到统计学，对异常用户的历史数据做统计分析，找出异于正常用户的特征实时性，风险事件的分析必须毫秒级响应，有些场景下需要尽快拦截，能够给用户止损挽回损失低误报，这需要人工风控经验，对各种场景风险阈值和评分的设置，需要长期不断的调整，所以灵活的规则引擎是很重要的支持对历史数据的回溯，能够发现以前的风险，或许能够找到一些特征供参考项目标签轻量级，可扩展，高性能的Java实时业务风控系统基于Springboot构建，配置文件能少则少使用drools规则引擎管理风控规则，原则上可以动态配置规则使用redis、mongodb做风控计算和事件储存，历史事件支持水平扩展基于Flink构建风控系统风控是一个很大的话题，涉及到规则引擎、NoSQLDB、CEP等等，本章主要讲一些风控的基本概念。在大数据侧，我们把风控划分成3×2的关系：2代表风控要么是基于规则的，要么是基于算法或模型的；3代表包括三种风控类型：事先风控、事中风控和事后风控。对于事中风控和事后风控来讲，端上的感知是异步的，对于事先风控来讲，端上的感知是同步的。对于事先风控这里稍做一些解释，事先风控是把已经训练好的模型或者把已经计算好的数据存在Redis、MongoDB等数据库中；一种方式是端上有类似Sidden、Groovy、Drools这样的规则引擎直接去Redis、MongoDB取数据来返回结果；另外一种方式是基于KubeflowKFserving，端上请求过来之后基于训练好的算法和模型返回结果。整体来讲这两种方式的时延都在200毫秒左右，可以作为一个同步的RPC或HTTP请求。对于Flink相关的大数据场景是一个异步的风控请求，它的异步时效性非常低，通常是一秒或者两秒。如果追求超低时延，则可以认为它是一种事中的风控，风控决策过程可以由机器介入处理。很常见的一种类型是用FlinkSQL做指标阈值的统计、用FlinkCEP做行为序列规则分析，还有一种是用TensorflowonFlink，在Tensorflow中进行算法描述，然后用Flink来执行Tensorflow规则的计算。

FlinkOnK8s实战课程分享下载，视频+源码+文档+虚拟机+软件包下载！！目前项目中用到Flink作为离线ETL处理构建相关的特征系统，而特征系统主要是为数据科学家、数据工程师、机器学习工程师去使用，用来去构建AI特征库，用来做模型的训练、用来做数据测试以及一些数据的预测及模型的在线服务，主要特征系统是从大数据Hbase、Hive以及关系型数据库去拉取相应的库表，存储到特征库表中，而本身K8S云原生也是相关的趋势，为什么flink要基于K8s做部署？主要有以下几个优势：容器环境容易部署、清理和重建：不像是虚拟环境以镜像进行分发部署起来对底层系统环境依赖小，所需要的包都很方便的集成到镜像中。更好的隔离性与安全性，应用部署以pod启动，pod之间相互独立，资源环境隔离后更安全。k8s集群能够利用好资源，机器学习、在线服务等许多任务都可以混合部署。云原生的趋势，丰富的k8s生态。编写K8s资源描述文件从FlinkonKubernetes的架构如上图所示，Flink任务在Kubernetes上运行的步骤有：（1）首先往Kubernetes集群提交了资源描述文件后，会启动Master和Worker的container。（2）MasterContainer中会启动FlinkMasterProcess，包含Flink-ContainerResourceManager、JobManager和ProgramRunner。（3）WorkerContainer会启动TaskManager，并向负责资源管理的ResourceManager进行注册，注册完成之后，由JobManager将具体的任务分给WorkerContainer，再由Container去执行。（4）需要说明的是，MasterContainer与WorkerContainer是用一个镜像启动的，只是启动参数不一样，如下图所示，两个deployment文件的镜像是同一个。首先，它会向K8sMaster申请创建FlinkConfigMap，在ConfigMap中提供了Flink集群运行所需要的配置，如：flink-conf.yaml和log4j.properties；其次，创建FlinkJobManager的service，通过service来打通TaskManager和JobManager之间的联通性；然后，创建FlinkJobmanager的Deployment，用来启动JobMaster，包含的组件有Dispatcher和Resourcemanager。最后，创建FlinkTaskManager的Deployment，用来启动TaskManager，因为Flink官方taskmanager-deployment.yaml示例中指定了2个副本，所以图中展示了2个TM节点

kubernetes，简称K8s，是用8代替名字中间的8个字符“ubernete”而成的缩写。是一个开源的，用于管理云平台中多个主机上的容器化的应用，Kubernetes的目标是让部署容器化的应用简单并且高效（powerful）,Kubernetes提供了应用部署，规划，更新，维护的一种机制。k8s通常被描述为一个容器编排（containerorchestration）平台。为了理解这个含义，让我们重新审视容器的作用，这有助于知道容器有哪些不足，以及k8s如何弥补这些不足。为什么我们喜欢使用容器？容器提供了一个轻量级的机制来隔离应用程序的环境。对于一个给定的应用程序，我们可以指定其配置和所需要安装的依赖，而不用担心其与同一台物理机上其他的应用程序发生冲突。我们将每个应用程序封装在容器镜像（containerimage）中，容器镜像可以可靠地运行在任何机器（只要机器有能力运行容器镜像）上，能够提供可移植的能力，即支持应用开发到部署的平滑过渡。此外，因为每个应用是独立的，不用担心环境冲突，所以在同一台物理机上可以部署多个容器，实现更高的资源（内存和CPU）利用率，最终降低成本。日志搜集当我们在FlinkonK8s上运行一个作业，有一个功能性问题无法回避，就是日志。如果是运行在YARN上，YARN会帮我们做这件事，例如在Container运行完成时，YARN会把日志收集起来传到HDFS，供后期查看。但是K8s并未提供日志搜集与存储，所以我们可以有很多选择去做日志（收集、展示）的事情。尤其是当作业因为异常导致POD退出，POD退出后日志会丢失，这将导致异常排查变得非常困难。首先往Kubernetes集群提交了资源描述文件后，会启动Master和Worker的container。MasterContainer中会启动FlinkMasterProcess，包含Flink-ContainerResourceManager、JobManager和ProgramRunner。WorkerContainer会启动TaskManager，并向负责资源管理的ResourceManager进行注册，注册完成之后，由JobManager将具体的任务分给Container，再由Container去执行。需要说明的是，MasterContainer与WorkerContainer是用一个镜像启动的，只是启动参数不一样。session模式FlinkSession集群作为长时间运行的KubernetesDeployment执行。你可以在一个Session集群上运行多个Flink作业。每个作业都需要在集群部署完成后提交到集群。Kubernetes中的FlinkSession集群部署至少包含三个组件：运行JobManager的部署TaskManagers池的部署暴露JobManager的REST和UI端口的服务

分享课程——FlinkOnK8s实战课程2023，附源码+文档+虚拟机+软件包下载。平台工程是一个在云原生时代重新焕发活力的旧概念，被视为解决云和集群扩张、资源浪费和成本失控问题的有效手段。容器和WebAssembly(WASM)提供了一个清晰的接口，使开发者可以自由选择他们喜欢的任何语言和框架（不同于JavaEE的限制），同时也便于核心团队进行平台标准的设定和治理。Kubernetes的清晰容器管理接口将开发和运营的关注点分离，从而提升效率和生产力。我知道许多开发者可能会抱怨："平台团队只是想再次控制我们使用的工具，他们总是阻碍我们，拖慢我们的进度，让我们感到困扰。"但我认为这次情况有所不同。为什么呢？因为有一个简单的约定，那就是只要能放进一个容器，它就能被部署。

Flink入门与实战-(大数据技术丛书)-汪明-清华大学出版社（epub格式，附阅读器安装程序）

Flink一线大厂实践经验.pdf

流式计算的代表：Storm、Flink、SparkStreaming.rar

作为新一代的开源流处理器，Flink是众多大数据处理框架中一颗冉冉升起的新星。它以同一种技术支持流处理和批处理，并能同时满足高吞吐、低延迟和容错的需求。本书由Flink项目核心成员执笔，系统阐释Flink的适用场景、设计理念、功能、用途和性能优势。-Flink的适用场景-流处理架构相较于批处理架构的优势-Flink中的时间概念-Flink的检查点机制-Flink的性能优势