## 流计算 1. 无限数据集 2. 有状态 3. 计算结束后数据丢弃 4. 事件驱动 流式计算是把数据汇集到有限的数据集中，就像是把源源不断的水流收集到各个桶中。 其中的一个原因是数据分析的需求，例如很多时候，人们关心的是“过去五分钟一共有多少次点击”、“过去一分钟最大的一笔订单是多少”之类的问题。 这种需求就是要对一个有限的数据集进行统计分析。 而从技术出发，像排序、聚合等数据操作，也都只是在有限的数据集上清晰的定义和成熟的算法。 ## 窗口 这种有限的数据集也被称作是窗口 什么时候去关闭并提交一个窗口，什么的情况下值得我们重新提交一个窗口。 ## 什么是事件驱动型应用？ 事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。 DataSource DataStream Window Transformation Operator 每隔5分钟输出最近一小时内同时在线用户数 用户平均使用速率 每隔5分钟用户行为分析 输入器连接数据源 => 数据流 想好输入输出，各个接口之间的交互情况 假设一条日志到达数据源，输入器通过订阅获取到这台数据，通知计算器进行计算，计算器存放计算结果，并在下条数据来临时更新结果 ## 输入器 * 负责连接数据源，从数据源中获取数据 * 当获取到数据时通知计算器 ## 计算器 * 接收输入器消息通知，进行业务逻辑计算，计算结果发送给输出器 * 维护时间窗口（保留一定时间范围内的数据），比如1小时 * 维护时间切片（这个时间范围内的数据按某个时间大小切片），比如5分钟 * 支持对时间窗口里的数据做聚合操作 * 通知输出器输出 ## 输出器 * 接受计算器的计算结果 * 将数据根据配置落地，推送，邮件日报 * 一份输出可能对应着多个计算器的计算结果 ## 调度器 * 读取配置文件 * 注入属性 interface Input { DataStream GetDataStream() } DataStream ds = Input.fromDataSource() // 输出器接入数据源，等待数据源中返回数据，一旦有数据，通知计算器计算 class Input { public void fromDataSource(DataStream ds) { ds.wait().then(function(msg) { // 当获取到一条消息时，通知计算器 msg.notifyAllCalculator() }) } } // 维护着一个时间范围内的数据 class Calculator { public Window window private void updateWindowTime(Message msg) {} public abstract compute(Message msg); public void onMessage(Message msg) { // 更新时间窗口的范围 updateWindowTime(msg) // 计算消息落到哪个时间窗口 } } ## 需求： 1. 正常日报内容输出 2. 实时监控各种数据 对于需要用到历史数据的指标怎么计算？ 3. 数据回流，利用老数据计算某个指标，工具 提供数据回流功能，将数据重新流入计算系统 4. 自定义实时监控 ## 各模块的依赖关系 计算器 -> 算法 存放在algorithm文件夹里的都是算法 ## 以下操作都需要什么改动 ### 增加输入器 输入器需要做抽象，适配不同数据源 数据源： * 消息队列 topic -> msg * 数据库 msg 不同的数据源的数据不一样，怎么搞定 ### 增加计算器 计算器的结构是稳定的，主要是算法不同，不同算法的输入输出都不一样 ## 增加输出器 ## core 1. 提供给输入器api，告知输入器要给我哪些数据（input接口） 2. 管理所有的算法列表，给算法传递数据 3. 算法的输出也可以是input接口 4. 4个列表（输出器列表，算法列表，值列表，输入器列表） ## 算法 1. 定义 2. 输入输出 ## 输出器 1. 定义我所需要的值 2. 输出方式 sys_login 计算在线用户数 数据源管理 DataSourceManager 给输入器的数据： MQ里的一条数据，SQL的一个结果集 DataSourceManager 订阅着所有的主题，主题消息一到，调用输入器的onMessage方法 subscribeEvent -> onEvent -> input.onMessage() 数据库 ## 计算器 ### 计算维度 如何计算各个唯独的维度，如何存储，怎么告诉 计算器如何表达我计算的东西是什么 ## 输出器 输出器怎么规定输出的内容，怎么表述 输出的 输出器告诉core，我需要哪些数据 日报输出器， 需要定时一天 ## 2. 怎么保证计算结果不丢失 * 写入文件？ * 定期写入数据库？ 3. 输出器应该是各项指标的混合，整理成需要的格式后按某种方式输出（邮件，入数据库） ## Core ### Blackboard * 保存需要实时显示的各种指标 ### EventEmitter * 提供给计算器计算完后发出事件 * 接受计算完成事件，通知黑板刷新，并按需要将计算结果流回core ## 数据提供器 ### 数据源 * kafka * Mysql ## 计算器 一个计算器只计算一个指标，多个计算器可以计算多个值 ### 时间窗口 计算器计算的结果属于某个时间范围，计算需要的中间数据以及计算的结构都只属于某个时间窗口，当窗口销毁时，应该重新计算 ### 临时数据 为了计算某个指标，计算器通常需要对收到的数据产生一份临时的中间数据 ### 输出数据 标记该计算器计算的结果的输出位置和格式，再对每一条新数据在原有的临时数据上做计算，最后将某个指标的值输出，每次计算完毕发出事件通知core， 计算器的输出是单个值 ## 输出器 一个输出器输出的内容是某个事件黑板上N个指标的整合输出，它只管去跟core要数据，要到数据后发出去就行嘞 ### 输出途径 * 数据库 * 邮件 ### 输出时间 什么时候去输出，日报按天，小时，分钟，秒 ### 输出结构 所有的指标需指定一定唯一名字，是否可以不要结构，只有单一的值 ## 新增用户 | 活跃用户 维度 * 天 * 小时 * 平台 * 版本 * 渠道 * 国家 // 日期 2019-07-31 // US新增 30 // US活跃 257 // US前日次留 21.21% // US连接用户数 86 // US连接次数 254 // US平均连接次数 2.95 // US矿机Crash 0 // US连通率 18.0% // US虚拟分配成功率 22.0% // US矿机连接成功率 93.0% // US数据获取成功率 90.0% // US矿机流量统计 // 上行:61.73G // 下行:78.14G // US用户平均使用流量 122.34MB // US平均上行速度 0.31 KB/S // US平均下行速度 17.77 KB/S // US用户平均使用时长 115.5 min // 注册成功率 73% // 登陆成功率 -- ## 存储过程的作用 * ot_user_login_hour 每小时计算活跃 * ot_user_conn_detail_hour 按小时计算连接用户前四项数据 * ot_queryClientSessionConn_hour 计算矿机连接成功率 * ot_business_line_data 更新业务线平台级联关系 * ot_user_keep 计算每日留存 select queryTime, count(DISTINCT username) nums from ot_user_login_hour t where 1=1 and queryTime >= ? and queryTime <= ? group by queryTime order by queryTime desc # 流式计算的难点 ## 时间 ## 有限的内存和无限的数据 ## 错误恢复 ## 内存计算 ## 需求： ### 正常日报内容输出 将现有利用数据库计算的一系列指标，迁移到流式计算框架中，目前有如下指标： * [x] US新增 * [x] US活跃 * [x] US前日次留 * [x] US连接用户数 * [x] US连接次数 * [x] US平均连接次数 * [x] US虚拟分配成功率 * [x] US矿机连接成功率 * [x] US数据获取成功率 * [x] 矿机crash次数 * [x] US流量统计 * [x] US上下行速度 * [x] US用户平均使用流量 * [x] US用户平均使用时长 * [x] 注册成功率 * [ ] 登陆成功率 ### 实时监控各种数据 - [ ] 可为某个指标设置阈值，超过时触发事件 ### 命令行工具 - [x] 可将数据库的历史数据重新流入计算框架，重新计算某指标 ### 自定义监控任务 如果启动后，考虑恢复数据 考虑定时任务，时间间隔 * 分钟 * 小时 * 天（按天定点输出） 当一个时间窗口关闭的时候，记录下状态值 { "time": { "begin": "", "end": "" }, "metrics": {} } 由于计算数据都是存放在计算器的temp和result里，都在某个时间段的时间窗口里， 所以如果需要恢复，就是恢复计算器的window，包括里面的temp和result 当重新启动的时候，收到第一条消息时，如果消息的时间仍在上次的时间窗口内，则恢复数据 那什么时候去保存数据 ### 黑板 黑板上的指标按时间段来分，以满足不同的输出器的输出需要 { "五分钟": { "begin": "", "end": "", "metrics": {} }, "十分钟": {}, "一小时": {}, "一天": {} } 对应算法计算的指标填入对应的时间段内 日报输出器查询黑板时，直接查询一天的数据，所以应该算法要对应计算到一天里去 时间窗口的区间怎么定义？ 时间窗口的时间以服务器的时间来定，不再以第一条日志的时间作为起始值 启动时获取时间戳作为起始时间，再获取下一个整点的时间作为结束时间 计算器的数据怎么保存，防止重启 每一分钟定时将window的数据写入磁盘 data "ActiveUserAlgorithm": { "begin": "", "end": "", "temp": "" "result": "" } 数据怎么保存？怎么恢复？如何将算法和数据对应起来，一个算法可能有多个实例 唯一标示：　算法名称 + 时间段 恢复时，查看算法对应时间段的数据是否存在，存在则恢复 场景: 总有一些指标是不适用与流式计算的,比如现在遇到的留存,对于这类指标,怎么计算方便 昨日留存率 = (昨日新增人 && 今天活跃人) / 昨日新增人数 像类似这种指标,需要用到历史数据,考虑范围加大,怎么算15日留存,难道必须等系统运行15天吗 是否通过某种方式,将需要用到的历史数据缓存到系统内 输出器: 当前所采集的数据只能是来源与黑板,如果可以来源与DB,问题是否可以解决 另外一个思路,多窗口是否能解决问题,那多窗口的时候,输出器怎么去哪里取数据, 数据在历史窗口里 写了一个bug,历史数据怎么处理 流入系统的数据类型 * 实时数据 * 无限数据 * 按原来的方式,固定窗口大小,时间取整,整5分钟,30分钟,一小时,一天 * 历史数据 * 有限数据 * 按数据的大小,数据输入有多少,就计算多少的量 今天完成的几个功能 1. 算法兼容两种数据类型 * 实时无限 要按时间切片来算数据 * 历史有限 可以直接以数据量来算 2. 留存 * 保留多窗口 * 或增加一种组件,接入历史数据 3. 完善工具 将昨天的新增用户通过 **某种方式** 导入进系统,就可以算了, 数据一定在数据库, 3. output的时候需要有一份外部数据,外部数据的更新需要一定的时间间隔 ./kafka-console-consumer.sh --bootstrap-server kafkas:9092 --topic user_register --from-beginning ./kafka-console-producer.sh --broker-list kafkas:9092 --topic user_register '{"accountname":"test@qq.com","action":"register","appname":"CoolLine","appversion":"1.6.xx","channelName":"googleplay","cityen":"","cityzh":"","continentsen":"","continentszh":"","countryen":"","countryZh":"美国","createTimestamp":1566376441,"deviceid":"edb3d8ce8df595ad","isPrivilegedUser":"true","line":"CoolLine","mail":"test@qq.com","pkgname":"cc.coolline.client","platform":"Android","provinceen":"","provincezh":"","pt":"2019-08-21","registerip":"10.244.2.64","registertime":"2019-08-21 16:34:01","registertype":"mail","userid":19348,"userName":"test@qq.com","userpwd":"86A8F132223D033619389988E663F6C2","userstate":0,"virtualCountryCode":"AF"}' 按任务计算 － 我去哪里取数据 查询一段时间，数据量大的问题，分批查，分批输入系统 ## 数据延迟到达 ## 数据到达了,被我丢弃未处理