flink1.17.0 集成kafka,并且计算

news2024/11/14 20:01:47

前言

flink是实时计算的重要集成组件,这里演示如何集成,并且使用一个小例子。例子是kafka输入消息,用逗号隔开,统计每个相同单词出现的次数,这么一个功能。


一、kafka环境准备

1.1 启动kafka

这里我使用的kafka版本是3.2.0,部署的方法可以参考,
kafka部署

cd kafka_2.13-3.2.0
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

启动后查看java进程是否存在,存在后执行下一步。

1.2 新建topic

新建一个专门用于flink消费topic

bin/kafka-topics.sh --create --topic flinkTest --bootstrap-server 192.168.184.129:9092

1.3 测试生产消费是否正常

生产端:

bin/kafka-console-producer.sh --topic flinkTest --bootstrap-server 192.168.184.129:9092

客户端:

bin/kafka-console-consumer.sh --topic flinkTest --from-beginning --bootstrap-server 192.168.184.129:9092

1.4 测试生产消费

在生产端输入aaa
在这里插入图片描述
查看客户端是否能消费到
在这里插入图片描述
可以看到客户端已经消费成功了,kafka环境准备好了。

二、flink集成kafka

2.1 pom文件修改

pom文件修改之前,先看看官网的指导依赖是什么样的,
这里我们使用的是datastream api去做,
flink1.17.0官方文档

在这里插入图片描述
这里说明了相关的依赖需要引入的依赖包的版本,还有使用kafka消费的时候需要引入的连接包版本
在这里插入图片描述
完整的pom引入依赖如下:

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.wh.flink</groupId>
    <artifactId>flink</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>flink</name>
    <!-- FIXME change it to the project's website -->
    <url>http://www.example.com</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.17.1</flink.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- Flink 依赖 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>


        <!-- Flink Kafka连接器的依赖 -->
<!--        <dependency>-->
<!--            <groupId>org.apache.flink</groupId>-->
<!--            <artifactId>flink-connector-kafka-0.11_2.11</artifactId>-->
<!--            <version>${flink.version}</version>-->
<!--        </dependency>-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>

        <!-- Flink 开发Scala需要导入以下依赖 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>${flink.version}</version>
            <!--<scope>provided</scope>-->
        </dependency>

        <!--<dependency>-->
        <!--<groupId>org.scala-lang</groupId>-->
        <!--<artifactId>scala-library</artifactId>-->
        <!--<version>2.11.12</version>-->
        <!--</dependency>-->

        <!-- log4j 和slf4j 包,如果在控制台不想看到日志,可以将下面的包注释掉-->
        <!--<dependency>-->
        <!--<groupId>org.slf4j</groupId>-->
        <!--<artifactId>slf4j-log4j12</artifactId>-->
        <!--<version>1.7.25</version>-->
        <!--<scope>test</scope>-->
        <!--</dependency>-->
        <!--<dependency>-->
        <!--<groupId>log4j</groupId>-->
        <!--<artifactId>log4j</artifactId>-->
        <!--<version>1.2.17</version>-->
        <!--</dependency>-->
        <!--<dependency>-->
        <!--<groupId>org.slf4j</groupId>-->
        <!--<artifactId>slf4j-api</artifactId>-->
        <!--<version>1.7.25</version>-->
        <!--</dependency>-->
        <!--<dependency>-->
        <!--<groupId>org.slf4j</groupId>-->
        <!--<artifactId>slf4j-nop</artifactId>-->
        <!--<version>1.7.25</version>-->
        <!--<scope>test</scope>-->
        <!--</dependency>-->
        <!--<dependency>-->
        <!--<groupId>org.slf4j</groupId>-->
        <!--<artifactId>slf4j-simple</artifactId>-->
        <!--<version>1.7.5</version>-->
        <!--</dependency>-->



    </dependencies>

    <build>
        <plugins>

            <!-- 在maven项目中既有java又有scala代码时配置 maven-scala-plugin 插件打包时可以将两类代码一起打包 -->
<!--            <plugin>-->
<!--                <groupId>org.scala-tools</groupId>-->
<!--                <artifactId>maven-scala-plugin</artifactId>-->
<!--                <version>2.15.2</version>-->
<!--                <executions>-->
<!--                    <execution>-->
<!--                        <goals>-->
<!--                            <goal>compile</goal>-->
<!--                            <goal>testCompile</goal>-->
<!--                        </goals>-->
<!--                    </execution>-->
<!--                </executions>-->
<!--            </plugin>-->

            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <!-- 设置false后是去掉 MySpark-1.0-SNAPSHOT-jar-with-dependencies.jar 后的 “-jar-with-dependencies” -->
                    <!--<appendAssemblyId>false</appendAssemblyId>-->
                    <archive>
                        <manifest>
                            <mainClass>com.hadoop.demo.service.flinkDemo.FlinkDemo</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>assembly</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

项目结构如图
在这里插入图片描述

2.2 代码编写

package com.hadoop.demo.service.flinkDemo;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.FlatMapIterator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
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 java.util.Arrays;
import java.util.Iterator;

public class FlinkDemo {




    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //新建kafka连接
        KafkaSource<String> kfkSource = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.184.129:9092")
                .setGroupId("flink")
                .setTopics("flinkTest")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        //添加到flink环境
        DataStreamSource<String> lines = env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "kafka source");
        //根据逗号分组
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = lines.flatMap(new FlatMapIterator<String, String>() {
            @Override
            public Iterator<String> flatMap(String s) throws Exception {
                return Arrays.asList(s.split(",")).iterator();
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return new Tuple2<>(s, 1);
            }
        });

        //统计每个单词的数量
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = map.keyBy(0).sum(1);
        sum.print();
        //System.out.println(sum.get);
        env.execute();
    }

}

2.3 maven打包在这里插入图片描述

点击打包按钮,这里注意要选择带依赖的jar包,否则会出现以下错误。

NoClassDefFoundError: org/apache/flink/connector/kafka/source/KafkaSource

三、测试

3.1启动 hadoop集群,启动flink集群

这里如果不知道怎么搭建这两个集群可以看我其他文章
hadoop集成flink

./hadoop.sh start
./bin/yarn-session.sh --detached

3.2 上传jar包到flink集群

在这里插入图片描述
上传后填写主类类名,点击提交
在这里插入图片描述

3.3 测试

点击后,可以看到执行job这里能看到在运行的job
在这里插入图片描述
点击运行的task
在这里插入图片描述
点击输出
在这里插入图片描述
这里可以看到输出内容,
在kafka消费端输入内容,
在这里插入图片描述
这里的jbs出现了4次,看下输出控制台,
在这里插入图片描述
可以看到这里依次累加了四次,说明统计生效了。


总结

这里只是做了一个简单的消费kafka的flink例子,消费成功后还可以通过sink发送出去,还可以用transform进行转换,这里后面再演示,如果不对的可以指出。

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

驾校驾考APP开发功能有哪些?

随着汽车成为越来越多人出行的代步工具之后&#xff0c;需要考驾照的人也是越来越多。小编记得我当初考驾照的时候还是抱着一个小本本每天刷题练习&#xff0c;小本本都快翻烂了。移动互联网的普及让驾考也开始走向线上&#xff0c;刷题、模拟、甚至是考试都可以通过驾考APP小程…

【csdn AI写作助手能帮助我们做什么呢?】

CSDN AI写作助手上线了&#xff01;InsCode AI 创作助手不仅能够帮助用户高效创作文章&#xff0c;而且能够作为对话式AI回答你想知道的问题。成倍提高生产力&#xff01; 一、你平时会使用这类AI工具吗&#xff1f;你对这类型的工具有什么看法&#xff1f; 提示&#xff1a;根…

Ubuntu离线安装Vsftp

这是资源包&#xff1a;(14条消息) unbuntu-vsftp.server-Linux文档类资源-CSDN文库 一、安装vsftp 将包解压&#xff0c;然后在解压报的目录下一键安装 dpkg -i *.deb // 安装所有 systemctl status vsftpd #查看运行状态 systemctl restart vsftpd #重新启动vsftp 二、…

【论文阅读公式推导1】连续体机器人的哈密尔顿动力学推导

推导了一下论文哈密尔顿原理的表达&#xff0c;原论文的计算公式是对的&#xff0c;记录一下。 Gravagne I A, Rahn C D, Walker I D. Good vibrations: a vibration damping setpoint controller for continuum robots[C]//Proceedings 2001 ICRA. IEEE International Confer…

[网站分享]

Element-ui Element - The worlds most popular Vue UI frameworkElement&#xff0c;一套为开发者、设计师和产品经理准备的基于 Vue 2.0 的桌面端组件库https://element.eleme.cn/#/zh-CN Vant Weapp Vant Weapp - 轻量、可靠的小程序 UI 组件库轻量、可靠的小程序 UI 组件…

没有数学基础可以学编程吗?

一、为什么学编程 这里我并不是问大家&#xff0c;是因为兴趣啊还是就业学编程。 而是&#xff0c;我想要学Python为了量化交易&#xff0c;或者我要处理表格。我想要学Java我就想自己建站。是否有这种非常明确的目标&#xff0c;有目标才能明确学习路线。 如果在这里&#…

大数据:HDFS操作的客户端big data tools和NFS

大数据&#xff1a;HDFS操作的客户端 2022找工作是学历、能力和运气的超强结合体&#xff0c;遇到寒冬&#xff0c;大厂不招人&#xff0c;可能很多算法学生都得去找开发&#xff0c;测开 测开的话&#xff0c;你就得学数据库&#xff0c;sql&#xff0c;oracle&#xff0c;尤…

Vue--》Vue3打造可扩展的项目管理系统后台的完整指南(三)

今天开始使用 vue3 ts 搭建一个项目管理的后台&#xff0c;因为文章会将项目的每一个地方代码的书写都会讲解到&#xff0c;所以本项目会分成好几篇文章进行讲解&#xff0c;我会在最后一篇文章中会将项目代码开源到我的GithHub上&#xff0c;大家可以自行去进行下载运行&…

Pytorch入门(二)神经网络的搭建

torch.nn中的nn全称为neural network,意思是神经网络&#xff0c;是torch中构建神经网络的模块。 文章目录 一、神经网络基本骨架二、认识卷积操作三、认识最大池化操作四、非线性激活五、线性层及其它层介绍六、简单的神经网络搭建七、简单的认识神经网络中的数值计算八、损失…

mmdetection训练coco数据集(继跑通后的一些工具使用)

&#xff08;仅做个人过程记录的笔记&#xff09; 1、生成中间件 可以选择评估方式 --eval &#xff0c;对于 COCO 数据集&#xff0c;可选 bbox 、segm、proposal 。可以得到result.bbox.json文件 生成pkl文件&#xff1a;faster_rcnn.pkl python tools/test.py config.py …

利用栈和队列共同解决迷宫问题

文章目录 什么是迷宫问题&#xff1f;如何解决迷宫问题&#xff1f;DFS&#xff08;深度优先搜索&#xff09;BFS&#xff08;广度优先搜索&#xff09; 总结 什么是迷宫问题&#xff1f; 迷宫问题是一道经典的算法问题&#xff0c;旨在寻找一条从起点到终点的最短路径。通常迷…

games101作业6

作业要求 Render() in Renderer.cpp: 将你的光线生成过程粘贴到此处&#xff0c;并且按照新框 架更新相应调用的格式。Triangle::getIntersection in Triangle.hpp: 将你的光线-三角形相交函数 粘贴到此处&#xff0c;并且按照新框架更新相应相交信息的格式。 在本次编程练习中…

删除排序链表中的重复元素(java)

删除链表中的重复元素 leetcode 83 题 删除链表中的重复元素解题思路代码链表专题 leetcode 83 题 删除链表中的重复元素 leetcode 83 题 – 跳转链接 给定一个已排序的链表的头 head &#xff0c; 删除所有重复的元素&#xff0c;使每个元素只出现一次 。返回 已排序的链表 。…

Python numpy - 数组与矩阵的创建&运算

数组array 一 数组的创建 &#xff08;至少两个数组&#xff09;创建随机整数数组 a 和 b import numpy as np a np.random.randint(10,size20).reshape(4,5) b np.random.randint(10,size20).reshape(4,5) 二 数组常用函数 数组常用函数 函数作用unique&#xff08;&…

Transformer part2

(179条消息) Transformer模型入门详解及代码实现_transformer模型代码-CSDN博客 transformer的encoder和decoder的差别 1. decoder包含两个 Multi-Head Attention 层。 decoder第一个 Multi-Head Attention 层采用了 Masked 操作。 为什么需要Mask处理 如何进行Mask处理 de…

【C#图解教程】第五章 类的基本概念

程序和类 类是一个能储存数据并执行代码的经过封装的数据结构&#xff0c;包含数据成员和函数成员&#xff0c;类内通常会包含逻辑上相关的数据和函数&#xff0c;所以类通常会代表真实世界或概念上的事物。 运行中的C#程序实质上是许多实例之间相互作用&#xff1a; 类的声…

ThingsBoard 前端项目内置部件开发

ThingsBoard 是目前 Github 上最流行的开源物联网平台&#xff08;12.8k Star&#xff09;&#xff0c;可以实现物联网项目的快速开发、管理和扩展&#xff0c;是中小微企业物联网平台的不二之选。 本文介绍如何在 ThingsBoard 前端项目中开发内置的菜单导航部件。 内置相关部…

clickhouse简介

文章目录 1&#xff1a;简介1.1&#xff1a;CH是什么&#xff1f;1.2&#xff1a;CH优势1.3&#xff1a;架构设计 2&#xff1a;CH接口3&#xff1a;CH引擎1&#xff1a;数据库引擎3.1.1:mysql引擎 2&#xff1a;表引擎3.2.1&#xff1a;MergeTree3.2.2&#xff1a;集成引擎1&a…

电子合同签署协议开源版系统开发

电子合同签署协议开源版系统开发 H5TP6mysqlphp 源码开源不加密 以下是电子合同系统可能包含的功能列表&#xff1a; 用户注册和登录&#xff1a;用户可以注册并登录系统&#xff0c;以便创建、签署和管理合同。合同创建&#xff1a;用户可以创建新合同&#xff0c;包括填写合…

Web的基本漏洞--SQL注入漏洞

目录 一、SQL注入介绍 1.SQL注入漏洞原理 2.SQL注入漏洞的类型 3.SQL注入漏洞识别 4.攻击方式 5.SQL盲注 时间盲注 布尔盲注 报错盲注 6.SQL注入漏洞的危害 7.SQL注入漏洞的防范措施 8.SQL注入漏洞的绕过 一、SQL注入介绍 1.SQL注入漏洞原理 Web程序输入的数据传…