内容简介:本系列文章将从热词统计需求讲起,讲解flink在实际项目中的应用、部署、容错。项目中的代码为scala所编写,没有用java的原因是scala语言看起来更加简练,同时与java语言的兼容性较好,可以直接引用java中的类。本篇文章会讲述热词统计最核心的一个功能实现,即消息传递过来的是搜索的单词,程序中进行统计。基于此我们定义传递的数据格式为定义好了数据格式之后,我们可以将其转换为scala中的Tuple,这里我们先利用fromElements api 创建离线数据用于调试
本系列文章将从热词统计需求讲起,讲解flink在实际项目中的应用、部署、容错。项目中的代码为scala所编写,没有用 java 的原因是scala语言看起来更加简练,同时与java语言的兼容性较好,可以直接引用java中的类。
定义数据格式
本篇文章会讲述热词统计最核心的一个功能实现,即消息传递过来的是搜索的单词,程序中进行统计。基于此我们定义传递的数据格式为 ${timestamp},${word}
,之所以有timestamp是为了标注搜索时间,用于修正flink中的watermask
离线数据模拟
定义好了数据格式之后,我们可以将其转换为scala中的Tuple,这里我们先利用fromElements api 创建离线数据用于调试
val currentTimeStamp = System.currentTimeMillis() val dataStream = env.fromElements( (currentTimeStamp,"word1"), (currentTimeStamp+1,"word2"), (currentTimeStamp+2,"word1"), (currentTimeStamp+3,"word3"), ) 复制代码
具体计算
我们要实现统计的功能,即求和,需要flink中的sum这个operator,通过[flink-1.8文档/操作符]发现,只有keyedStream这个对象才有sum方法,所以在执行sum之前得先调用keyBy方法将dataStream转换为keyedStream。接下来是这两个方法具体的传参,我们有两个字段,时间戳和单词,要对单词进行keyBy操作需传入1(下标从0开始),但是sum方法要传什么呢?这个时候我们发现我们的对象中没有可以用来求和的,所以我们需要将数据结构从 ${timestamp},${word}
转换为 ${timestamp},${word},${count}
,这就得使用map方法了,其中count为1。经过如上分析,有如下代码
dataStream .map(x => (x._1, x._2, 1)) .keyBy(1) .sum(2) .print() 复制代码
为便于调试,最后的结果输出为到控制台即为
6> (1556680015646,word2,1) 12> (1556680015648,word3,1) 10> (1556680015647,word1,1) 10> (1556680015647,word1,2)
输出结果中开头的数字即为线程的标示,这里我们看出,word2和word3分别是不同的线程,而word1则为同一个线程
疑问解答
- 为什么这里关于word1的统计有两次输出呢? 因为这里是实时流式计算,来一条数据,各个operator就会为它计算一次,所以最后都有一次对应的输出
- 那什么时候输出的个数会变少呢? 在使用window的时候,多个数据会按照一定的规则(根据个数或者根据时间段)聚集在一个window中,然后每一个window就只有一个输出结果
小技巧
- 这里我们使用的scala语言来编写flink项目,需要注意的是要在pom.xml文件中加上如下代码,使其能够支持scala编译
<plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <executions> <!-- Run scala compiler in the process-resources phase, so that dependencies on scala classes can be resolved later in the (Java) compile phase --> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>compile</goal> </goals> </execution> <!-- Run scala compiler in the process-test-resources phase, so that dependencies on scala classes can be resolved later in the (Java) test-compile phase --> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testCompile</goal> </goals> </execution> </executions> <configuration> <jvmArgs> <jvmArg>-Xms128m</jvmArg> <jvmArg>-Xmx512m</jvmArg> </jvmArgs> </configuration> </plugin> </plugins> 复制代码
- 在项目的引用中,需要引用scala后缀的,如
<properties> <scala.binary.version>2.11</scala.binary.version> <flink.version>1.8.0</flink.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> </dependencies> 复制代码
- 文件中的引用,在你输入
StreamExecutionEnvironment
的时候,Intellij会提示你引入org.apache.flink.streaming.api.scala. StreamExecutionEnvironment
,但是当你调用env.fromElements()
方法的时候你会编译器会有类型错误的提示
import org.apache.flink.streaming.api.scala._
就可以了
以上所述就是小编给大家介绍的《Flink 热词统计(1): 基础功能实现》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!
猜你喜欢:- 微信小程序数据统计和错误统计的实现
- 基于jieba库实现中文词频统计
- Java8中使用stream进行分组统计和普通实现的分组统计的性能对比
- golang实现Elasticsearch做短信查询统计
- C#黔驴技巧之实现统计结果排名
- python实现微信群友统计器
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
测出转化率:营销优化的科学与艺术
【美】高尔德(Goward,C.) / 谭磊、唐捷译 / 电子工业出版社 / 2014-10-1 / 68.00元
本书作者通过已成功实现大幅提升转化率的案例,展示了大量以营销为核心的电子商务网站的测试设计方法及转化优化方案。书中作者强调了测试及优化思维的重要性,并就实现方法做了详细讲解。 通过本书,读者将学到如何能够在网站遇到发展和收入瓶颈时,测试出存在的问题并找到解决方案;如何可以深入地了解客户需求,并以此为基础优化网站,使其达到提升转化率的目的;如何提升网站的竞争优势,把在线营销渠道变成高效的转化通......一起来看看 《测出转化率:营销优化的科学与艺术》 这本书的介绍吧!