Flink实战六_直播礼物统计

接上文:Flink实战五_状态机制

1、需求背景

现在网络直播平台非常火爆,在斗鱼这样的网络直播间,经常可以看到这样的总榜排名,体现了主播的人气值。

人气值计算规则:用户发送1条弹幕互动,赠送1个荧光棒免费道具、100个免费鱼丸、亲密度礼物等行为,均可为主播贡献1点及以上人气值。

我们就以这个人气值日榜为例,来设计一个Flink的计算程序。

在这里插入图片描述
对于人气值日榜这样的功能,可以理解为是一个典型的流式计算的场景,强调的是数据的实时处理。因为在这个场景下,必须要及时的累计用户的送礼物数据,才能形成你追我赶的实时效果,提升用户的参与体验。这个场景下的实时性,虽然不要求每一条数据都及时响应,但是整体的数据延迟还是要尽量缩短的。

这种场景下,使用Flink进行流批统一的计算,感觉就非常合适。

2、数据流程设计

在确定了使用Flink进行计算后,首先就需要设计出数据的上下游流程,进行简单的方案可行性评估。

对于数据上游,我们这个人气值日榜统计的业务场景,数据来源自然就是粉丝们的打赏行为。一方面整个平台的打赏行为的数据量是非常大的,另一方面这些打赏行为涉及到账户操作,所以他的作用,更大的是体现在人气值榜功能以外的其他业务过程中。基于这两方面考虑,自然就会想到使用kafka来进行削峰以及解耦。而Flink在DataStream/DataSet API和 Table API&SQL 两个部分都对kafka提供了连接器实现,所以用kafka作为数据接入是可行的。

而对于数据下游,其实可以想象,最终计算出来的数据,最为重要的是要强调查询的灵活性以及时效性,这样才能支持页面的快速查询。如果考虑查询的时效性,HBase和ElasticSearch都是比较理想的大数据存储引擎。但是如果考虑到查询的灵活性,就会想到ElasticSearch会相比HBase更适合。因为我们统计出来的这些粉丝人气值度的结果,不光可以作为每个直播间人气值榜的排名,也应该可以作为以后平台主播年度排名等其他业务场景的数据来源。如果想要兼顾这些查询场景,使用HBase就会对Rowkey产生大量的侵入,而Elasticsearch可以根据任意字段快速查询,就比较有优势。 另外,从官方文档中可以查到,对于HBase,Flink只提供了Table API&SQL 模块的connector支持,而DataStream/DataSet API中没有提供支持,而ElasticSearch则支持更为全面。当然,这跟HBase的具体场景是有关联的,但是也可以从另一个角度认为,使用ElasticSearch的可行性更高。

这样,就初步确定了 kafka-> Flink -> ElasticSearch 这样的大致数据流程。这
也是在实际开发中非常典型的一个组合方式。后续就可以着手搭建kafka集群以及ElasticSearch+Kibana的集群了。搭建的过程就略过了。

确定数据的基础结构
这一步主要是确定入口数据和出口数据的结构。只要这两个数据结构确定了,那
么应用程序模块和大数据计算模块就可以分开进行开发了。是双方主要的解耦方
式。

在数据入口处,可以定义这样的简化的数据结构:

public static class GiftRecord{
private String hostId; //主播ID
private String fansId; //粉丝ID
private long giftCount; //礼物数量
private String giftTime; //送礼物时间。时间格式 yyyy-MM-DD HH:mm:SS
.....
}

在kafka中,确定使用gift作为Topic,MQ的消息格式为 #{hostId},#{fansId},#{giftCount},#{giftTime} 这样的字符串。

在数据出口处,可以定义ES中这样简化的索引结构:

-- 贡献日榜索引
PUT daygiftanalyze
{
"mappings":{"properties": {"windowEnd":{"type": "long"},"hostId": {"type": "keyword"},"fansId": {"type": "keyword"},"giftCount":{"type": "long"}}}
}

这样,一个简单的设计方案就形成了。应用程序只需要在粉丝发送礼物时往kafka中同步一条消息记录,然后从ES中查询主播的人气值日榜和人气值周榜数据即可。而我们也可以模拟数据格式进行开发了。

3、应用实现

人气值日榜:
基础数据结构:

public static class GiftRecord{private String hostId; //主播IDprivate String fansId; //粉丝IDprivate long giftCount; //礼物数量private String giftTime; //送礼物时间。时间格式 yyyy-MM-DD HH:mm:SS.....
}

在kafka中,确定使用gift作为Topic,MQ的消息格式为 #{hostId},#{fansId},#{giftCount},#{giftTime} 这样的字符串。

ES索引:

PUT daygiftanalyze
{"mappings": {"properties": {"windowEnd": {"type": "long"},"hostId": {"type": "keyword"},"fansId": {"type": "keyword"},"giftCount": {"type": "long"}}}
}

然后运行Flink程序,com.flink.project.flink.DayGiftAna,从kafka中读取数
据。测试数据见giftrecord.txt。计算程序会及时将十分钟内的粉丝礼物统计都存入到ES当中。

giftrecord.txt如下:

1001,3001,100,2021-09-15 15:15:10
1001,3002,321,2021-09-15 15:17:14
1001,3003,234,2021-09-15 15:16:24
1001,3004,15,2021-09-15 15:17:13
1001,3005,264,2021-09-15 15:18:14
1001,3006,678,2021-09-15 15:17:54
1001,3007,123,2021-09-15 15:19:22
1001,3008,422,2021-09-15 15:18:37
1001,3009,566,2021-09-15 15:22:43
1001,3001,76,2021-09-15 15:21:28
1001,3001,88,2021-09-15 15:26:28
1001,3007,168,2021-09-15 15:32:29
1001,3002,157,2021-09-15 15:28:56
1001,3009,567,2021-09-15 15:27:32
1001,3004,145,2021-09-15 15:30:26
1001,3003,1656,2021-09-15 15:31:19
1001,3005,543,2021-09-15 15:36:49
1001,3001,864,2021-09-15 15:38:26
1001,3001,548,2021-09-15 15:45:10
1001,3007,359,2021-09-15 15:52:39
1001,3008,394,2021-09-15 15:59:48

com.flink.project.flink.DayGiftAna,如下:


import com.roy.flink.project.fansgift.FansGiftResult;
import com.roy.flink.project.fansgift.GiftRecord;
import org.apache.commons.lang.SystemUtils;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;/*** @desc 贡献日榜计算程序*/
public class DayGiftAna {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.getConfig().setAutoWatermarkInterval(1000L); //BoundedOutOfOrdernessWatermarks定时提交Watermark的间隔
//        env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop01:8020/dayGiftAna"));// Checkpoint存储到文件if(SystemUtils.IS_OS_WINDOWS){env.setStateBackend(new FsStateBackend("file:///D:/flink_file"));}else{// linuxenv.setStateBackend(new FsStateBackend("file:///home/file_file"));}//使用Socket测试。env.setParallelism(1);final DataStreamSource<String> dataStream = env.socketTextStream("10.86.97.206", 7777);final SingleOutputStreamOperator<FansGiftResult> fansGiftResult = dataStream.map((MapFunction<String, GiftRecord>) value -> {final String[] valueSplit = value.split(",");//SimpleDateFormat 多线程不安全。SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");final long giftTime = sdf.parse(valueSplit[3]).getTime();return new GiftRecord(valueSplit[0], valueSplit[1], Integer.parseInt(valueSplit[2]), giftTime);}).assignTimestampsAndWatermarks(WatermarkStrategy.<GiftRecord>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner((SerializableTimestampAssigner<GiftRecord>) (element, recordTimestamp) -> element.getGiftTime()))
//          .keyBy((KeySelector<GiftRecord, String>) value -> value.getHostId() + "_" + value.getFansId()) //按照HostId_FansId分组.keyBy((KeySelector<GiftRecord, String>) value -> value.getHostId()) //按照HostId分组.window(TumblingEventTimeWindows.of(Time.seconds(10)))
//                .allowedLateness(Time.seconds(2)).aggregate(new WinodwGiftRecordAgg(), new AllWindowGiftRecordAgg());//打印结果测试fansGiftResult.print("fansGiftResult");env.execute("DayGiftAna");}//在每个子任务中将窗口期内的礼物进行累计合并//增加状态后端。private static class WinodwGiftRecordAgg implements AggregateFunction<GiftRecord, Long, Long> {@Overridepublic Long createAccumulator() {return 0L;}@Overridepublic Long add(GiftRecord value, Long accumulator) {Long res = accumulator + value.getGiftCount();return res;}@Overridepublic Long getResult(Long accumulator) {return accumulator;}@Overridepublic Long merge(Long a, Long b) {return a + b;}}//对窗口期内的所有子任务进行窗口聚合操作。private static class AllWindowGiftRecordAgg extends RichWindowFunction<Long, FansGiftResult, String, TimeWindow> {ValueState<FansGiftResult> state;@Overridepublic void apply(String s, TimeWindow window, java.lang.Iterable<Long> input, Collector<FansGiftResult> out) throws Exception {final String[] splitKey = s.split("_");String hostId = splitKey[0];String fansId ="";if(splitKey.length>1){fansId=splitKey[1];}final Long giftCount = input.iterator().next();final long windowEnd = window.getEnd();final FansGiftResult fansGiftResult = new FansGiftResult(hostId, fansId, giftCount, windowEnd);out.collect(fansGiftResult);state.update(fansGiftResult);}@Overridepublic void open(Configuration parameters) throws Exception {final ValueStateDescriptor<FansGiftResult> stateDescriptor = new ValueStateDescriptor<>("WinodwGiftRecordAgg", TypeInformation.of(new TypeHint<FansGiftResult>() {}));state = this.getRuntimeContext().getState(stateDescriptor);}}
}

FansGiftResult,代码如下:

public class FansGiftResult {private String hostId;private String fansId;private long giftCount;private long windowEnd;public FansGiftResult() {}public FansGiftResult(String hostId, String fansId, long giftCount, long windowEnd) {this.hostId = hostId;this.fansId = fansId;this.giftCount = giftCount;this.windowEnd = windowEnd;}@Overridepublic String toString() {if(fansId!=null && fansId.length()>0){return "FansGiftResult{" +"hostId='" + hostId + '\'' +", fansId='" + fansId + '\'' +", giftCount=" + giftCount +", windowEnd=" + windowEnd +'}';}else{return "FansGiftResult{" +"hostId='" + hostId + '\'' +", giftCount=" + giftCount +", windowEnd=" + windowEnd +'}';}}public String getHostId() {return hostId;}public void setHostId(String hostId) {this.hostId = hostId;}public String getFansId() {return fansId;}public void setFansId(String fansId) {this.fansId = fansId;}public long getGiftCount() {return giftCount;}public void setGiftCount(long giftCount) {this.giftCount = giftCount;}public long getWindowEnd() {return windowEnd;}public void setWindowEnd(long windowEnd) {this.windowEnd = windowEnd;}
}

GiftRecord,代码如下:


public class GiftRecord {private String hostId; //主播IDprivate String fansId; //粉丝IDprivate int giftCount; //礼物数量private long giftTime; //送礼物时间。原始时间格式 yyyy-MM-DD HH:mm:ss,ssspublic GiftRecord() {}public GiftRecord(String hostId, String fansId, int giftCount, long giftTime) {this.hostId = hostId;this.fansId = fansId;this.giftCount = giftCount;this.giftTime = giftTime;}public String getHostId() {return hostId;}public void setHostId(String hostId) {this.hostId = hostId;}public String getFansId() {return fansId;}public void setFansId(String fansId) {this.fansId = fansId;}public int getGiftCount() {return giftCount;}public void setGiftCount(int giftCount) {this.giftCount = giftCount;}public long getGiftTime() {return giftTime;}public void setGiftTime(long giftTime) {this.giftTime = giftTime;}@Overridepublic String toString() {return "GiftRecord{" +"hostId='" + hostId + '\'' +", fansId='" + fansId + '\'' +", giftCount=" + giftCount +", giftTime='" + giftTime + '\'' +'}';}
}

ES查询语句:

GET daygiftanalyze/_search
{"query": {"bool": {"must": [{"range": {"windowEnd": {"gte": 1631635200000,"lte": 1631721600000}}},{"match": {"hostId": "1001"}}]}},"aggs": {"groupByFans": {"terms": {"field": "fansId","size": 3,"order": {"giftCount": "desc"}},"aggs": {"giftCount": {"sum": {"field": "giftCount"}}}}}
}

ES中的查询结果:
在这里插入图片描述
直播应用就可以根据这个查询结果组织客户端查询代码,最终实现日榜排名的功能。

4、实现效果分析

具体的计算方案参见示例代码,这里就不多做分析了。这里只分析一下在实现过程中需要注意的几个重要的问题:

  • 时间语义分析
    对于网络直播这样的场景,从下午六点到第二天早上六点才是一天的高峰期,所以,在进行统计时,将每一天的统计时间定义为从早上六点到第二天早上六点,这样就能尽量保持高峰期的完整性。很多跟娱乐相关的场景,比如网络游戏,也大都是以这样的范围来定义一天,而不是传统意义上的从0点到24点。

  • 并行度优化
    可以直接使用Flink的开窗机制,待一周的数据收集完整了之后,一次性向ES中输出统计结果,这种场景下要注意累计器的持久化,以及计算程序出错后的重启恢复机制。

  • 后续改进方式
    状态后端、而对于人气值日榜的计算,就不能等一天的数据收集齐了再计算了。这时是有两种解决方案,一种是完全的流处理方式。也就是每来一条数据就往ES中更新结果。另一中方式是采用小批量的流处理方式。以五分钟为单位,将数据拆分成一个一个小窗
    口来进行处理。显然后一种方式对数据处理的压力会比较小一点。虽然数据量会更
    多,但是ES的存储以及快速查询能力可以比较好的弥补数据量的问题。也因此,在
    设计ES数据机构时,将人气值日榜的文档结构设计成了一个一个的小范围。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/253932.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

海外云手机的核心优势

随着5G时代的到来&#xff0c;云计算产业正处于高速发展的时期&#xff0c;为海外云手机的问世创造了一个可信任的背景。在资源有限且需求不断增加的时代&#xff0c;将硬件设备集中在云端&#xff0c;降低个人用户的硬件消耗&#xff0c;同时提升性能&#xff0c;这一点单单就…

2024牛客寒假算法基础集训营1

A 找dfs这三个字符即可 #include<bits/stdc.h> #define IOS ios::sync_with_stdio(0);cin.tie(0);cout.tie(0); #define endl \nusing namespace std;typedef pair<int, int> PII; typedef long long ll;const int N 55;int n; char s[N];void solve() {cin >…

LeetCode Python - 1.两数之和

文章目录 题目答案运行结果 题目 给定一个整数数组 nums 和一个整数目标值 target&#xff0c;请你在该数组中找出 和为目标值 target 的那 两个 整数&#xff0c;并返回它们的数组下标。 你可以假设每种输入只会对应一个答案。但是&#xff0c;数组中同一个元素在答案里不能…

【网站项目】031网络游戏公司官方平台

&#x1f64a;作者简介&#xff1a;拥有多年开发工作经验&#xff0c;分享技术代码帮助学生学习&#xff0c;独立完成自己的项目或者毕业设计。 代码可以私聊博主获取。&#x1f339;赠送计算机毕业设计600个选题excel文件&#xff0c;帮助大学选题。赠送开题报告模板&#xff…

博途PLC报警字FC(字寄存器按位访问)

博途PLC的字寄存器按位访问和拆分,请查看下面文章链接: https://rxxw-control.blog.csdn.net/article/details/121727057https://rxxw-control.blog.csdn.net/article/details/121727057西门子触摸屏报警都是以字为地址访问,所以离散报警信号我们需要将其组合为报警字输出,…

async 与 await(JavaScript)

目录捏 前言一、async二、await三、使用方法总结 前言 async / await 是 ES2017(ES8) 提出的基于 Promise 解决异步的最终方案。上一篇文章介绍了 回调地狱 与 Promise&#xff08;JavaScript&#xff09;&#xff0c;因为 Promise 的编程模型依然充斥着大量的 then 方法&#…

【芯片设计- RTL 数字逻辑设计入门 11.1 -- 状态机实现 移位运算与乘法 1】

文章目录 移位运算与乘法状态机简介SystemVerilog中的测试平台VCS 波形仿真 阻塞赋值和非阻塞赋值有限状态机&#xff08;FSM&#xff09;与无限状态机的区别 本篇文章接着上篇文章【芯片设计- RTL 数字逻辑设计入门 11 – 移位运算与乘法】 继续介绍&#xff0c;这里使用状态机…

MyBatis之动态代理实现增删改查以及MyBatis-config.xml中读取DB信息文件和SQL中JavaBean别名配置

MyBatis之环境搭建以及实现增删改查 前言实现步骤1. 编写MyBatis-config.xml配置文件2. 编写Mapper.xml文件&#xff08;增删改查SQL文&#xff09;3. 定义PeronMapper接口4. 编写测试类1. 执行步骤2. 代码实例3. 运行log 开发环境构造图总结 前言 上一篇文章&#xff0c;我们…

C++后端开发之Sylar学习二:配置VSCode远程连接Ubuntu开发

C后端开发之Sylar学习二&#xff1a;配置VSCode远程连接Ubuntu开发 没错&#xff0c;我不能像大佬那样直接在Ubuntu上面用Vim手搓代码&#xff0c;只能在本地配置一下VSCode远程连接Ubuntu进行开发咯&#xff01; 本篇主要是讲解了VSCode如何配置ssh连接Ubuntu&#xff0c;还有…

Oracle笔记-为表空间新增磁盘(ORA-01691)

如下报错&#xff1a; 原因是Oracle表空间满了&#xff0c;最好是新增一个存储盘。 #查XXX命名空间目前占用了多大的空间 select FILE_NAME,BYTES/1024/1024 from dba_data_files where tablespace_name XXXX #这里的FILE_NAME能查到DBF的存储位置#将对应的datafile设置为30g…

#免费 苹果M系芯片Macbook电脑MacOS使用Bash脚本写入(读写)NTFS硬盘教程

Mac电脑苹果芯片读写NTFS硬盘bash脚本 &#xff08;ntfs.sh脚本内容在本文最后面&#xff09; ntfs.sh脚本可以将Mac系统(苹果M系芯片)上的NTFS硬盘改成可读写的挂载方式&#xff0c;从而可以直接往NTFS硬盘写入数据。此脚本免费&#xff0c;使用过程中无需下载任何收费软件。…

vue教程-介绍与使用

vue介绍 介绍 是一套用于构建用户界面的渐进式框架。与其它大型框架不同的是&#xff0c;Vue 被设计为可以自底向上逐层应用。Vue 的核心库只关注视图层&#xff0c;不仅易于上手&#xff0c;还便于与第三方库或既有项目整合。 安装 最简单的例子就是&#xff0c;创建一个htm…

云上未来:探索云计算的技术变革与应用趋势

一、云计算的起源和演进 1.1 早期计算模型 在探讨云计算的起源和演进之前&#xff0c;理解早期的计算模型对于构建全面的视角至关重要。早期计算模型的发展奠定了云计算的基础&#xff0c;为其演进提供了技术和理念的支撑。 1.1.1 集中式计算模型 在计算技术的早期阶段&…

JVM Java虚拟机入门指南

文章目录 为什么学习JVMJVM的执行流程JVM的组成部分类加载运行时数据区本地方法接口执行引擎 垃圾回收什么样的对象是垃圾呢内存溢出和内存泄漏定位垃圾的方法对象的finalization机制垃圾回收算法分代回收垃圾回收器 JVM调优参数JVM调优工具Java内存泄漏排查思路CPU飙高排查方案…

私有化部署一个吃豆人小游戏

目录 效果 安装步骤 1.安装并启动httpd 2.下载代码 3.启动httpd 使用 效果 安装步骤 1.安装并启动httpd yum -y install httpd 2.下载代码 进入目录 cd /var/www/html/ 下载 git clone https://gitee.com/WangZhe168_admin/pacman-canvas.git 3.启动httpd syste…

c++阶梯之类与对象(中)< 续集 >

前文&#xff1a; c阶梯之类与对象&#xff08;上&#xff09;-CSDN博客 c阶梯之类与对象&#xff08;中&#xff09;-CSDN博客 前言&#xff1a; 在上文中&#xff0c;我们学习了类的六个默认成员函数之构造&#xff0c;析构与拷贝构造函数&#xff0c;接下来我们来看看剩下…

常用的EasyExcel表格处理-2(动态合并、自适应宽高)

EasyExcel官网&#xff1a;点击查看 1、动态合并单元格 此处主要根据自定义处理类ExcelFillCellMergeStrategy进行处理&#xff0c;具体内容可看代码注释。 1.1 前端调用controller PostMapping("/download/template")public void toDoExport(HttpServletResponse…

c#string方法对比

字符串的截取匹配操作在开发中非常常见&#xff0c;比如下面这个示例&#xff1a;我要匹配查找出来字符串数组中以“abc”开头的字符串并打印&#xff0c;我下面分别用了两种方式实现&#xff0c;代码如下&#xff1a; using System; namespace ConsoleApp23{ class Progra…

Android开发 button 按钮点击两次 响应onclick方法

问题 Android开发 button 按钮点击两次 响应onclick方法 详细问题 笔者xml代码 <!-- 一个按钮 --> <Button android:id"id/button1" android:layout_width"wrap_conten…

Rust 第一个rust程序Hello Rust️

文章目录 前言一、vscode 安装rust相关插件二、Cargo New三、vscode调试rustLLDB 前言 Rust学习系列。今天就让我们掌握第一个rust程序。Hello Rust &#x1f980;️。 在上一篇文章我们在macOS成功安装了rust。 一、vscode 安装rust相关插件 以下是一些常用的 Rust 开发插件…