Windows环境下Flink消费Kafka实现热词统计

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
注册配置 MSE Nacos/ZooKeeper,182元/月
任务调度 XXL-JOB 版免费试用,400 元额度,开发版规格
简介: 前言碎语昨天博主写了《windows环境下flink入门demo实例》实现了官方提供的最简单的单词计数功能,今天升级下,将数据源从socket流换成生产级的消息队列kafka来完成一样的单词计数功能。

前言碎语

昨天博主写了《windows环境下flink入门demo实例》实现了官方提供的最简单的单词计数功能,今天升级下,将数据源从socket流换成生产级的消息队列kafka来完成一样的单词计数功能。本文实现的重点主要有两个部分,一是kafka环境的搭建,二是如何使用官方提供的flink-connector-kafka_2.12来消费kafka消息,其他的逻辑部分和上文类似。

进入正题

本篇博文涉及到的软件工具以及下载地址:

Apache Flink :https://flink.apache.org/downloads.html ,请下载最新版1.7.x,选择单机版本

kafka:http://kafka.apache.org/downloads ,请下载最新的2.1.0

第一步:安装kafka,并验证 

从上面的下载地址选择二进制包下载后是个压缩包,解压后的目录如下:

进入bin\windows下,找到kafka-server-start.bat和zookeeper-server-start.bat。配置文件在config目录下,主要配置一些日志和kafka server和zookeeper,都默认就好。如果你本地已经有zk的环境,就可以忽略zk,不然按照下面的步骤执行即可。

1. 启动zk服务

执行:zookeeper-server-start.bat ..\..\config\zookeeper.properties

2.启动kafka服务

执行:kafka-server-start.bat ..\..\config\server.properties

3.创建test主题

执行:kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

4.查看上一步的主题是否创建成功,成功的话控制台会输出test

执行:kafka-topics.bat --list --zookeeper localhost:2181

5.订阅test主题消息

执行:kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

6.发布消息

执行:kafka-console-producer.bat --broker-list localhost:9092 --topic test

以上步骤成功后,我们需要验证下是否都成功了。在第六条指令的窗口中输入abc。如果在第5个指令窗口输出了就代表kafka环境ok了。然后可以关掉第5个指令窗口,下面就让Flink来消费kafka的消息

第二步:编写消费kafka消息的Flink job

基础步骤参考《windows环境下flink入门demo实例》一文。唯一的区别就是因为要消费kafka中的数据,所以需要引入一个kafka连接器,官方已提供到maven仓库中,引入最新版本即可,如下:

<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.12</artifactId>
<version>1.7.1</version>

然后新建一个KafkaToFlink类 ,代码逻辑和昨天的一样,都是从一段字符串中统计每个词语出现的次数,这个场景比较像我们的热搜关键字,我标题简化为热词统计了。主要的代码如下:

/**
* Created by kl on 2019/1/30.
* Content :消费kafka数据
*/
public class KafkaToFlink {

public static void main(String\[\] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  env.enableCheckpointing(5000);

    /\*\*
     \* 这里主要配置KafkaConsumerConfig需要的属性,如:
     \* --bootstrap.servers localhost:9092 --topic test --group.id test-consumer-group
     */
    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<String>(parameterTool.getRequired("topic"), new SimpleStringSchema(), parameterTool.getProperties()));
    DataStream<WordWithCount> windowCounts = dataStream.rebalance().flatMap(new FlatMapFunction<String, WordWithCount>() {
        public void flatMap(String value, Collector<WordWithCount> out) {
            System.out.println("接收到kafka数据:" + value);
            for (String word : value.split("\\\s")) {
                out.collect(new WordWithCount(word, 1L));
            }
        }
    }).keyBy("word")
            .timeWindow(Time.seconds(2))
            .reduce(new ReduceFunction<WordWithCount>() {
                public WordWithCount reduce(WordWithCount a, WordWithCount b) {
                    return new WordWithCount(a.word, a.count + b.count);
                }
            });
    windowCounts.print().setParallelism(1);
    env.execute("KafkaToFlink");
}

}

注意下这个地方:ParameterTool.fromArgs(args);我们所有的关于KafkaConsumerConfig的配置,都是通过启动参数传入的,然后Flink提供了一个从args中获取参数的工具类。这里需要配置的就三个信息,和我们在命令窗口创建订阅一样的参数即可

第三步:验证Flink job是否符合预期

将应用打成jar包后通过Flink web上传到Flink Server。然后,找到你提交的job,输入如下的启动参数,提交submit即可:

成功运行的job的页面如下图,如果下图框框中的指标一直在转圈圈,那么很有可能是因为你运行了其他的job,导致Available Task Slots不够用了。

默认的Flink的Slots配置是1,当出现任务插槽不够用时,上图圈圈转一会就会失败,然后打开job manager 点击log就可以看到job因为没有可用的任务插槽而失败了。

org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not allocate all requires slots within timeout of 300000 ms. Slots required: 2, slots allocated: 0

at org.apache.flink.runtime.executiongraph.ExecutionGraph.lambda$scheduleEager$3(ExecutionGraph.java:991)
at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:870)
at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:852)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
at org.apache.flink.runtime.concurrent.FutureUtils$ResultConjunctFuture.handleCompletedFuture(FutureUtils.java:535)
at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:772)
at akka.dispatch.OnComplete.internal(Future.scala:258)
at akka.dispatch.OnComplete.internal(Future.scala:256)
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:186)
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:183)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:83)
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:534)
at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:20)
at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:18)
at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436)
at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91)
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

上面的问题可以通过修改conf/flink-conf.yaml中的taskmanager.numberOfTaskSlots来设置,具体指单个TaskManager可以运行的并行操作员或用户功能实例的数量。如果此值大于1,则单个TaskManager将获取函数或运算符的多个实例。这样,TaskManager可以使用多个CPU内核,但同时,可用内存在不同的操作员或功能实例之间划分。此值通常与TaskManager的计算机具有的物理CPU核心数成比例(例如,等于核心数,或核心数的一半)。当然,如果你修改了配置文件,Flink Server是需要重启的。重启成功后,可以在大盘看到,如下图箭头:

一切就绪后,在kafka-console-producer窗口中输入字符串回车,就会在flink job窗口中看到相关的信息了,效果前文一样,如图:

文末结语

本文算昨天hello wrod入门程序的升级版,实现了消费kafka中的消息来统计热词的功能。后面生产环境也打算使用kafka来传递从mysql binlog中心解析到的消息,算是一个生产实例的敲门砖吧。正如博主昨天所说的,落地的过程肯定会有很多问题,像上面的taskmanager.numberOfTaskSlots的设置。后面会继续将我们落地过程中的问题记录下来,欢迎关注凯京科技一起交流。

相关文章
|
5月前
|
XML 存储 搜索推荐
Omnissa Dynamic Environment Manager 2503 - 个性化动态 Windows 桌面环境管理
Omnissa Dynamic Environment Manager 2503 - 个性化动态 Windows 桌面环境管理
101 7
Omnissa Dynamic Environment Manager 2503 - 个性化动态 Windows 桌面环境管理
|
5月前
|
Ubuntu 数据库 虚拟化
Windows 环境下 Odoo 安装保姆级教程
本教程详细介绍了在 Windows 系统上通过虚拟机部署 Odoo 的完整流程。首先确认硬件需求,确保 CPU、内存和磁盘空间满足最低配置;接着安装 VMware Workstation Pro 并创建 Ubuntu 虚拟机,配置桥接网络以实现主机与虚拟机的通信;随后借助微聚云快速安装预配置好的 Odoo 环境,简化复杂环境搭建;最后通过浏览器访问虚拟机 IP,完成 Odoo 数据库初始化及基础设置。整个过程清晰易懂,适合新手快速上手 Odoo 部署。
708 4
|
6月前
|
存储 运维 监控
提升Windows Server环境安全性:ADAudit Plus的五大关键优势
在Windows Server环境中,内置的安全审计工具虽有用,但存在专业门槛高、耗时及功能缺失等问题。第三方工具ADAudit Plus应运而生,其五大优势包括:日志聚合、关键活动检测、定制化报告、灵活安全配置和长期日志保留,有效提升系统监控与合规能力。选择ADAudit Plus,助力企业更高效应对审计挑战,强化安全性。
153 2
|
Linux C++ Windows
【Azure 应用服务】Azure App Service(Windows)环境中如何让.NET应用调用SAP NetWeaver RFC函数
【Azure 应用服务】Azure App Service(Windows)环境中如何让.NET应用调用SAP NetWeaver RFC函数
145 0
【Azure 应用服务】Azure App Service(Windows)环境中如何让.NET应用调用SAP NetWeaver RFC函数
|
9月前
|
消息中间件 Java Kafka
【手把手教你Linux环境下快速搭建Kafka集群】内含脚本分发教程,实现一键部署多个Kafka节点
本文介绍了Kafka集群的搭建过程,涵盖从虚拟机安装到集群测试的详细步骤。首先规划了集群架构,包括三台Kafka Broker节点,并说明了分布式环境下的服务进程配置。接着,通过VMware导入模板机并克隆出三台虚拟机(kafka-broker1、kafka-broker2、kafka-broker3),分别设置IP地址和主机名。随后,依次安装JDK、ZooKeeper和Kafka,并配置相应的环境变量与启动脚本,确保各组件能正常运行。最后,通过编写启停脚本简化集群的操作流程,并对集群进行测试,验证其功能完整性。整个过程强调了自动化脚本的应用,提高了部署效率。
2130 1
【手把手教你Linux环境下快速搭建Kafka集群】内含脚本分发教程,实现一键部署多个Kafka节点
|
9月前
|
弹性计算 开发框架 安全
基于云效 Windows 构建环境和 Nuget 制品仓库进行 .Net 应用开发
本文将基于云效 Flow 流水线 Windows 构建环境和云效 Packages Nuget 制品仓库手把手教你如何开发并部署一个 .NET 应用,从环境搭建到实战应用发布的详细教程,帮助你掌握 .NET 开发的核心技能。
|
10月前
|
消息中间件 资源调度 关系型数据库
如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理
本文介绍了如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理。主要内容包括安装Debezium、配置Kafka Connect、创建Flink任务以及启动任务的具体步骤,为构建实时数据管道提供了详细指导。
502 9
|
10月前
|
Dart 搜索推荐 IDE
Windows下Zed编辑器配置Dart环境
本文介绍了Dart编程语言及其主要框架Flutter的优势,并推荐使用轻量级编辑器Zed进行Dart开发。详细步骤包括Dart环境的安装与配置,Zed编辑器的安装与个性化设置,以及如何在Zed中编写并运行Dart的HelloWorld程序。通过自定义任务实现Dart文件的快速运行,提高了开发效率。
|
11月前
|
应用服务中间件 Shell PHP
windows系统配置nginx环境运行pbootcms访问首页直接404的问题
windows系统配置nginx环境运行pbootcms访问首页直接404的问题
|
11月前
|
Oracle 关系型数据库 MySQL
Mysql(1)—简介及Windows环境下载安装
MySQL 是一个流行的关系型数据库管理系统(RDBMS),基于 SQL 进行操作。它由瑞典 MySQL AB 公司开发,后被 Sun Microsystems 收购,现为 Oracle 产品。MySQL 是最广泛使用的开源数据库之一,适用于 Web 应用程序、数据仓库和企业应用。
158 2

热门文章

最新文章