k8s 搭建基于session模式的flink集群

1.flink集群搭建

不废话直接上代码,都是基于官网的,在此记录一下 Kubernetes | Apache Flink

flink-configuration-configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:name: flink-configlabels:app: flink
data:flink-conf.yaml: |+jobmanager.rpc.address: flink-jobmanagertaskmanager.numberOfTaskSlots: 2blob.server.port: 6124jobmanager.rpc.port: 6123taskmanager.rpc.port: 6122jobmanager.memory.process.size: 1600mtaskmanager.memory.process.size: 1728mparallelism.default: 2    log4j-console.properties: |+# This affects logging for both user code and FlinkrootLogger.level = INFOrootLogger.appenderRef.console.ref = ConsoleAppenderrootLogger.appenderRef.rolling.ref = RollingFileAppender# Uncomment this if you want to _only_ change Flink's logging#logger.flink.name = org.apache.flink#logger.flink.level = INFO# The following lines keep the log level of common libraries/connectors on# log level INFO. The root logger does not override this. You have to manually# change the log levels here.logger.pekko.name = org.apache.pekkologger.pekko.level = INFOlogger.kafka.name= org.apache.kafkalogger.kafka.level = INFOlogger.hadoop.name = org.apache.hadooplogger.hadoop.level = INFOlogger.zookeeper.name = org.apache.zookeeperlogger.zookeeper.level = INFO# Log all infos to the consoleappender.console.name = ConsoleAppenderappender.console.type = CONSOLEappender.console.layout.type = PatternLayoutappender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n# Log all infos in the given rolling fileappender.rolling.name = RollingFileAppenderappender.rolling.type = RollingFileappender.rolling.append = falseappender.rolling.fileName = ${sys:log.file}appender.rolling.filePattern = ${sys:log.file}.%iappender.rolling.layout.type = PatternLayoutappender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%nappender.rolling.policies.type = Policiesappender.rolling.policies.size.type = SizeBasedTriggeringPolicyappender.rolling.policies.size.size=100MBappender.rolling.strategy.type = DefaultRolloverStrategyappender.rolling.strategy.max = 10# Suppress the irrelevant (wrong) warnings from the Netty channel handlerlogger.netty.name = org.jboss.netty.channel.DefaultChannelPipelinelogger.netty.level = OFF    

jobmanager-service.yaml Optional service, which is only necessary for non-HA mode.

apiVersion: v1
kind: Service
metadata:name: flink-jobmanager
spec:type: ClusterIPports:- name: rpcport: 6123- name: blob-serverport: 6124- name: webuiport: 8081selector:app: flinkcomponent: jobmanager

Session cluster resource definitions #

jobmanager-session-deployment-non-ha.yaml

apiVersion: apps/v1
kind: Deployment
metadata:name: flink-jobmanager
spec:replicas: 1selector:matchLabels:app: flinkcomponent: jobmanagertemplate:metadata:labels:app: flinkcomponent: jobmanagerspec:containers:- name: jobmanagerimage: apache/flink:latestargs: ["jobmanager"]ports:- containerPort: 6123name: rpc- containerPort: 6124name: blob-server- containerPort: 8081name: webuilivenessProbe:tcpSocket:port: 6123initialDelaySeconds: 30periodSeconds: 60volumeMounts:- name: flink-config-volumemountPath: /opt/flink/confsecurityContext:runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessaryvolumes:- name: flink-config-volumeconfigMap:name: flink-configitems:- key: flink-conf.yamlpath: flink-conf.yaml- key: log4j-console.propertiespath: log4j-console.properties

taskmanager-session-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:name: flink-taskmanager
spec:replicas: 2selector:matchLabels:app: flinkcomponent: taskmanagertemplate:metadata:labels:app: flinkcomponent: taskmanagerspec:containers:- name: taskmanagerimage: apache/flink:latestargs: ["taskmanager"]ports:- containerPort: 6122name: rpclivenessProbe:tcpSocket:port: 6122initialDelaySeconds: 30periodSeconds: 60volumeMounts:- name: flink-config-volumemountPath: /opt/flink/conf/securityContext:runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessaryvolumes:- name: flink-config-volumeconfigMap:name: flink-configitems:- key: flink-conf.yamlpath: flink-conf.yaml- key: log4j-console.propertiespath: log4j-console.properties

 kubectl apply -f xxx.yaml 或者 kubectl apply -f ./flink  flink为文件夹,存放的是以上这几个.yaml文件

为flink的ui界面添加nodeport即可外部访问

2. demo代码测试

创建一个maven工程,pom.xml引入依赖:

<?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"><parent><artifactId>test-platform</artifactId><groupId>com.test</groupId><version>2.0.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>flink-demo</artifactId><properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>1.17.0</flink.version><log4j.version>2.20.0</log4j.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId><scope>compile</scope><version>${log4j.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-api</artifactId><scope>compile</scope><version>${log4j.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><scope>compile</scope><version>${log4j.version}</version></dependency></dependencies></project>

log4j2.xml:

<?xml version="1.0" encoding="UTF-8"?>
<configuration monitorInterval="5"><Properties><property name="LOG_PATTERN" value="%date{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n" /><!-- LOG_LEVEL 配置你需要的日志输出级别       --><property name="LOG_LEVEL" value="INFO" /></Properties><appenders><console name="Console" target="SYSTEM_OUT"><PatternLayout pattern="${LOG_PATTERN}"/><ThresholdFilter level="${LOG_LEVEL}" onMatch="ACCEPT" onMismatch="DENY"/></console></appenders><loggers><root level="${LOG_LEVEL}"><appender-ref ref="Console"/></root></loggers></configuration>

计数代码:

package com.test.flink;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class WordCountUnboundStreamDemo {public static void main(String[] args) throws Exception {// TODO 1.创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
//                3, // 尝试重启的次数
//                Time.of(10, TimeUnit.SECONDS) // 间隔
//        ));// TODO 2.读取数据DataStreamSource<String> lineDS = env.socketTextStream("192.168.0.28", 7777);// TODO 3.处理数据: 切分、转换、分组、聚合// TODO 3.1 切分、转换SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = lineDS //<输入类型, 输出类型>.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {// 按照 空格 切分String[] words = value.split(" ");for (String word : words) {// 转换成 二元组 (word,1)Tuple2<String, Integer> wordsAndOne = Tuple2.of(word, 1);// 通过 采集器 向下游发送数据out.collect(wordsAndOne);}}});// TODO 3.2 分组KeyedStream<Tuple2<String, Integer>, String> wordAndOneKS = wordAndOneDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}});// TODO 3.3 聚合SingleOutputStreamOperator<Tuple2<String, Integer>> sumDS = wordAndOneKS.sum(1);// TODO 4.输出数据sumDS.print("接收到的数据=======").setParallelism(1);// TODO 5.执行:类似 sparkstreaming最后 ssc.start()env.execute(sumDS.getClass().getSimpleName());}}

打成jar包导入flink dashboard:

在另一台机器上运行 nc -lk -p 7777,如果出现连接拒绝,查看是否放开端口号

k8s查看读取到的数据

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

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

相关文章

postgresql-条件表达式

postgresql-条件表达式 简单Case表达式搜索Case表达式缩写函数总结 简单Case表达式 select e.first_name , e.last_name , e.department_id , case e.department_id when 90 then 管理when 60 then 开发else 其他end as "部门" from cps.public.employees e ;-- 统…

Vue笔记

第一章&#xff1a;Vue环境搭建 1.搭建Vue环境 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>Title</title><!-- 1.引入Vue.js--><script src"1.vue.js"></scr…

软件生命周期及流程

软件生命周期&#xff1a; 软件生命周期(SDLC&#xff0c;Systems Development Life Cycle)是软件开始研制到最终被废弃不用所经历的各个阶段. 需求分析阶段--输出需求规格说明书&#xff08;原型图&#xff09; 测试介入的晚--回溯成本高 敏捷开发模型&#xff1a; 从1990年…

STM32CUBEMX_创建时间片轮询架构的软件框架

STM32CUBEMX_创建时间片轮询架构的软件框架 说明&#xff1a; 1、这种架构避免在更新STM32CUBEMX配置后把用户代码清除掉 2、利用这种时间片的架构可以使得代码架构清晰易于维护 创建步骤&#xff1a; 1、使用STM32CUBEMX创建基础工程 2、新建用户代码目录 3、构建基础的代码框…

OpenLdap +PhpLdapAdmin + Grafana docker-compose部署安装

目录 一、OpenLdap介绍 二、PhpLdapAdmin介绍 三、使用docker-compose进行安装 1. docker-compose.yml 2. grafana配置文件 3. provisioning 四、安装openldap、phpldapadmin、grafana 五、配置OpenLDAP 1. 登陆PhpLdapAdmin web管理 2. 需要注意的细节 内容介绍参考…

Java作业3

1.下面代码的运行结果是&#xff08;C&#xff09; public static void main(String[] args){String s;System.out.println("s"s);}A.代码编程成功&#xff0c;并输出”s” B.代码编译成功&#xff0c;并输出”snull” C.由于String s没有初始化&#xff0c;代码不…

python基础运用例子

python基础运用例子 1、⼀⾏代码交换 a , b &#xff1a;a, b b, a2、⼀⾏代码反转列表 l[::-1]3、合并两个字典 res {**dict1, **dict2}**操作符合并两个字典for循环合并dict(a, **b) 的方式dict(a.items() b.items()) 的方式dict.update(other_dict) 的方式 4、⼀⾏代码列…

【多尺度双域引导网络:Pan-sharpening】

Multi-Scale Dual-Domain Guidance Network for Pan-sharpening &#xff08;用于泛锐化的多尺度双域引导网络&#xff09; 全色锐化的目标是在纹理丰富的全色图像的指导下&#xff0c;通过超分辨低空间分辨率多光谱图像&#xff08;LRMS&#xff09;的对应物产生高空间分辨率…

语音特征提取与预处理

导入相关包 import librosa import librosa.display import soundfile as sf import numpy as np import matplotlib.pyplot as plt from playsound import playsound 语音读取与显示 file_path test1.wav data, fs librosa.load(file_path, srNone, monoTrue) librosa.d…

数学建模--整数规划匈牙利算法的Python实现

目录 1.算法流程简介 2.算法核心代码 3.算法效果展示 1.算法流程简介 #整数规划模型--匈牙利算法求解 """ 整数规划模型及概念&#xff1a;规划问题的数学模型一般由三个因素构成 决策变量 目标函数 约束条件&#xff1b;线性规划即以线性函数为目标函数&a…

2024腾讯校招后端面试真题汇总及其解答(三)

21【算法题】反转链表 题目: 给定单链表的头节点 head ,请反转链表,并返回反转后的链表的头节点。 示例 1: 输入:head = [1,2,3,4,5] 输出:[5,4,3,2,1]示例 2: 输入:head = [1,2] 输出:[2,1]示例 3: 输入:head = [] 输出:[]提示: 链表中节点的数目范围是 [0, 5…

GPT转换工具:轻松将MBR转换为GPT磁盘

为什么需要将MBR转换为GPT&#xff1f; 众所周知&#xff0c;Windows 11已经发布很长时间了。在此期间&#xff0c;许多老用户已经从Windows 10升级到Windows 11。但有些用户仍在运行Windows 10。对于那些想要升级到Win 11的用户来说&#xff0c;他们可能不确定Win 11应该使…

LeetCode 热题 100——找到字符串中所有字母异位词(滑动窗口)

题目链接 力扣&#xff08;LeetCode&#xff09;官网 - 全球极客挚爱的技术成长平台 题目解析 该题目的意思简而言之就是说&#xff0c;从s字符串中寻找与p字符串含有相同字符(次数和种类均相同)的子串&#xff0c;并且将他们的首字符下标集合进数组中进行返回。 滑动窗口解…

大数据Flink(七十三):SQL的滚动窗口(TUMBLE)

文章目录 SQL的滚动窗口(TUMBLE) SQL的滚动窗口(TUMBLE) 滚动窗口定义:滚动窗口将每个元素指定给指定窗口大小的窗口。滚动窗口具有固定大小,且不重叠。例如,指定一个大小为 5 分钟的滚动窗口。在这种情况下,Flink 将每隔 5 分钟开启一个新的窗口,其中每一条数都会划…

如何使用蚂蚁集团自动化混沌工程 ChaosMeta 做 OceanBase 攻防演练?

当前&#xff0c;业界主流的混沌工程项目基本只关注如何制造故障的问题&#xff0c;而经常做演练相关工作的工程师应该明白&#xff0c;每次演练时还会遇到以下痛点&#xff1a; 检测当前环境是否符合演练预设条件&#xff08;演练准入&#xff09;&#xff1b; 业务流量是否满…

Vue基础1:生命周期汇总(vue2)

Description 生命周期图&#xff1a; 可以理解vue生命周期就是指vue实例从创建到销毁的过程&#xff0c;在vue中分为9个阶段&#xff1a;创建前/后&#xff0c;载入前/后&#xff0c;更新前/后&#xff0c;销毁前/后&#xff0c;其他&#xff1b;常用的有&#xff1a;created&…

C#常用多线程(线程同步,事件触发,信号量,互斥锁,共享内存,消息队列)

using System; using System.Threading; using System.Windows.Forms; using UtilForm.Util;namespace UtilForm {// 线程同步&#xff0c;事件触发&#xff0c;信号量&#xff0c;互斥锁&#xff0c;共享内存&#xff0c;消息队列public partial class frmUIThread : Form{ Sy…

autojs修改顶部标题栏颜色

顶部标题栏的名字是statusBarColor 不是toolbar。难怪我搜索半天搜不到 修改之后变成这样了 代码如下&#xff1a; "ui"; importClass(android.view.View); importClass(android.graphics.Color); ui.statusBarColor(Color.parseColor("#ffffff")); ui.…

一个月能做什么?成长感悟分享

一个月做了什么&#xff1f; 八月做了些什么&#xff1f; 单词打卡 第一件事情就是单词打卡 英语很差的我&#xff0c;一样继续打卡&#xff0c;今天是第736天 当你还在纠结扇贝和不背、可可英语哪一个好的时候&#xff0c;别人已经同时使用了 当你还在咨询学编程、敲代码需…

Pygame中Trivia游戏解析6-1

1 Trivia游戏简介 Trivia的含义是“智力测验比赛中的各种知识”。Trivia游戏类似智力竞赛&#xff0c;由电脑出题&#xff0c;玩家进行作答&#xff0c;之后电脑对玩家的答案进行判断&#xff0c;给出结果并进行评分。该游戏的界面如图1所示。 图1 Trivia游戏界面 2 游戏流程 …