-
springboot集成flink并发布flink集群端运行
背景：近期项目需要，引入flink，研究了下flink，步步踩坑终于可以单独运行，也可发布到集群运行，记录下踩坑点。开发环境：idea+springboot(2.3.5.RELEASSE)+kafka(2.8.1)+mysql(8.0.26)。废话不多说，直接上可执行代码。
以下代码实现了某个时间间隔，设备不上传数据，判断为离线的逻辑
一、项目application创建
/** * flink任务提交application * * @author wangfenglei */ @SpringBootApplication(scanBasePackages = {"com.wfl.firefighting.flink","com.wfl.firefighting.util"}) public class DataAnalysisFlinkApplication { public static void main(String[] args) { SpringApplication.run(DataAnalysisFlinkApplication.class, args); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
二、设备状态计算主体，从kafka接收数据，然后通过KeyedProcessFunction函数进行计算，然后把离线设备输出到mysql sink，更新设备状态
/** * 从kafka读取数据，计算设备状态为离线后写入mysql * * @author wangfenglei */ @Component @ConditionalOnProperty(name = "customer.flink.cal-device-status", havingValue = "true", matchIfMissing = false) public class DeviceDataKafkaSource { private static final Logger log = LoggerFactory.getLogger(CalDeviceOfflineFunction.class); @Value("${spring.kafka.bootstrap-servers:localhost:9092}") private String kafkaServer; @Value("${spring.kafka.properties.sasl.jaas.config}") private String loginConfig; @Value("${customer.flink.cal-device-status-topic}") private String topic; @Autowired private ApplicationContext applicationContext; /** * 执行方法 * * @throws Exception 异常 */ @PostConstruct public void execute() throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000); env.setParallelism(1); Properties properties = new Properties(); //kafka的节点的IP或者hostName，多个使用逗号分隔 properties.setProperty("bootstrap.servers", kafkaServer); //kafka的消费者的group.id properties.setProperty("group.id", "data-nanlysis-flink-devicestatus"); //设置kafka安全认证机制为PLAIN properties.setProperty("sasl.mechanism", "PLAIN"); //设置kafka安全认证协议为SASL_PLAINTEXT properties.setProperty("security.protocol", "SASL_PLAINTEXT"); //设置kafka登录验证用户名和密码 properties.setProperty("sasl.jaas.config", loginConfig); FlinkKafkaConsumermyConsumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties); DataStream stream = env.addSource(myConsumer); stream.print().setParallelism(1); DataStream deviceStatus = stream //进行转换只获取设备序列码 .map(data -> CommonConstant.GSON.fromJson(data, MsgData.class).getDevSn()) //按照设备序列码分组 .keyBy(new KeySelector - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
说明：
1、通过@ConditionalOnProperty开关形式控制程序是否执行，后续此模块可以开发多个flink执行任务，通过开关的形式提交需要的job
2、通过springboot的@PostConstruct注解，让项目application启动时，自动执行job
3、用Thread线程执行任务提交，否则application启动时会一直flink执行中
4、日志打印，需要使用slf4j，跟flink自己依赖jar包打印日志保持一致，如此在flink集群执行时可以打印日志
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;private static final Logger log = LoggerFactory.getLogger(CalDeviceOfflineFunction.class);
5、kafka连接开启了登录验证，配置见application.yml。kafka登录验证server端配置见官网资料，后续有时间写个文章记录下
三、设备离线计算
/** * KeyedProcessFunction 为每个设备序列码维护一个state，并且会把间隔时间内(事件时间)内没有更新的设备序列码输出： * 对于每条记录，CalDeviceOfflineFunction 修改最后的时间戳。 * 该函数还会在间隔时间内调用回调（事件时间）。 * 每次调用回调时，都会检查存储的最后修改时间与回调的事件时间时间戳，如果匹配则发送设备序列码（即在间隔时间内没有更新,表示没有设备数据上传） * * @author wangfenglei */ @Component @ConditionalOnProperty(name = "customer.flink.cal-device-status", havingValue = "true", matchIfMissing = false) public class CalDeviceOfflineFunction extends KeyedProcessFunctiondeviceState; /** * 定时任务执行时间 */ private ValueState timerState; @Autowired private DeviceService deviceService; @Override public void open(Configuration parameters) throws Exception { deviceState = getRuntimeContext().getState(new ValueStateDescriptor<>("deviceState", DeviceLastDataTimestamp.class)); timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timerState", Long.class)); } /** * 每条数据执行过程 * * @param value 输入数据 * @param ctx 环境 * @param out 输出数据 * @throws Exception 异常 */ @Override public void processElement(String value, Context ctx, Collector out) throws Exception { log.info("++++++++++++++fink recevice deviceSn={}", value); // 查看当前计数 DeviceLastDataTimestamp current = deviceState.value(); if (current == null) { current = new DeviceLastDataTimestamp(); current.key = value; current.lastDataTime = ctx.timestamp(); } Long currentTimerState = timerState.value(); if (null == currentTimerState) { //初始值设置为-1 timerState.update(-1L); } if (-1 != timerState.value()) { //删除原先定时任务，然后重新注册新的定时任务 ctx.timerService().deleteProcessingTimeTimer(timerState.value()); } long interval = deviceService.getDeviceOfflineInterval(value); // 设置状态的时间戳为记录的事件时间时间戳 current.lastDataTime = ctx.timestamp(); //设置判断离线时间间隔 current.interval = interval; // 状态回写 deviceState.update(current); //更新定时任务执行时间 timerState.update(current.lastDataTime + interval); //注册新的定时任务 ctx.timerService().registerProcessingTimeTimer(current.lastDataTime + interval); } /** * 定时器触发后执行的方法 * * @param timestamp 这个时间戳代表的是该定时器的触发时间 * @param ctx 定时器环境类 * @param out 输出 * @throws Exception 异常 */ @Override public void onTimer( long timestamp, OnTimerContext ctx, Collector out) throws Exception { // 取得该设备状态的State状态 DeviceLastDataTimestamp result = deviceState.value(); // timestamp是定时器触发时间，如果等于最后一次更新时间+离线间隔时间，就表示这十秒内没有收到过该设备报文了 if (timestamp == result.lastDataTime + result.interval) { // 发送 out.collect(result.key); // 打印数据，用于核对是否符合预期 log.info("==================" + result.key + " is offline"); } } /** * 设备最后上传数据时间戳数据类 */ class DeviceLastDataTimestamp { public String key; public long lastDataTime; public long interval; } } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
四、 更新设备离线状态
/** * 向mysql写入数据 * * @author wangfenglei */ @Component @ConditionalOnProperty(name = "customer.flink.cal-device-status", havingValue = "true", matchIfMissing = false) public class DeviceStatusSink extends RichSinkFunction{ private static final Logger log = LoggerFactory.getLogger(DeviceStatusSink.class); @Value("${spring.datasource.dynamic.datasource.master.url}") private String datasoureUrl; @Value("${spring.datasource.dynamic.datasource.master.username}") private String userName; @Value("${spring.datasource.dynamic.datasource.master.password}") private String password; @Value("${spring.datasource.dynamic.datasource.master.driver-class-name}") private String driverClass; private Connection conn = null; private PreparedStatement ps = null; @Override public void open(Configuration parameters) throws Exception { //加载驱动,开启连接 try { Class.forName(driverClass); conn = DriverManager.getConnection(datasoureUrl, userName, password); } catch (Exception e) { e.printStackTrace(); } } @Override public void invoke(String deviceSn, Context context) { try { String sql = "update biz_device t set t.status=2 where t.dev_sn=?"; ps = conn.prepareStatement(sql); ps.setString(1, deviceSn); ps.executeUpdate(); log.info("update biz_device t set t.status=2 where t.dev_sn={}", deviceSn); } catch (Exception e) { e.printStackTrace(); } } /** * 结束任务,关闭连接 * * @throws Exception */ @Override public void close() throws Exception { if (conn != null) { conn.close(); } if (ps != null) { ps.close(); } } } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
五、application.yml配置
server: port: 8099 spring: autoconfigure: exclude: com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure datasource: druid: stat-view-servlet: enabled: true loginUsername: admin loginPassword: 123456 allow: web-stat-filter: enabled: true dynamic: druid: # 全局druid参数，绝大部分值和默认保持一致。(现已支持的参数如下,不清楚含义不要乱设置) # 连接池的配置信息 # 初始化大小，最小，最大 initial-size: 5 min-idle: 5 maxActive: 20 # 配置获取连接等待超时的时间 maxWait: 60000 # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒 timeBetweenEvictionRunsMillis: 60000 # 配置一个连接在池中最小生存的时间，单位是毫秒 minEvictableIdleTimeMillis: 300000 validationQuery: SELECT 1 FROM DUAL testWhileIdle: true testOnBorrow: false testOnReturn: false # 打开PSCache，并且指定每个连接上PSCache的大小 poolPreparedStatements: true maxPoolPreparedStatementPerConnectionSize: 20 # 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙 filters: stat,wall,slf4j # 通过connectProperties属性来打开mergeSql功能；慢SQL记录 connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000 datasource: master: url: jdbc:mysql://127.0.0.1:3306/fire?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai username: root password: root driver-class-name: com.mysql.cj.jdbc.Driver kafka: bootstrap-servers: 127.0.0.1:9092 # 指定kafka 代理地址，可以多个 producer: # 生产者 retries: 1 # 设置大于0的值，则客户端会将发送失败的记录重新发送 # 每次批量发送消息的数量 batch-size: 16384 buffer-memory: 33554432 # 指定消息key和消息体的编解码方式 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer #修改最大向kafka推送消息大小 properties: max.request.size: 52428800 consumer: group-id: data-analysis-flink #手动提交offset保证数据一定被消费 enable-auto-commit: false #指定从最近地方开始消费(earliest) auto-offset-reset: latest #消费者组 #group-id: dev properties: #服务端没有收到心跳超时时间，设置长点以防调试时超时 session: timeout: ms: 60000 heartbeat: interval: ms: 30000 security: protocol: SASL_PLAINTEXT sasl: mechanism: PLAIN jaas: config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="root" password="root";' #自定义配置 customer: #flink相关配置 flink: #是否开启设置状态计算 cal-device-status: true cal-device-status-topic: device-upload-data- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
六、pom.xml文件
4.0.0 com.wfl.firefighting data-analysis 1.0.0 com.wfl.firefighting data-analysis-flink 1.0.0 jar org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-logging org.springframework.boot spring-boot-starter-test test com.wfl.firefighting data-analysis-service 1.0.0 com.wfl.firefighting data-analysis-model 1.0.0 io.github.openfeign feign-httpclient 10.10.1 com.alibaba druid-spring-boot-starter 1.1.22 com.baomidou dynamic-datasource-spring-boot-starter 2.5.4 mysql mysql-connector-java 8.0.20 org.apache.flink flink-java 1.13.1 org.apache.flink flink-streaming-java_2.11 1.13.1 org.apache.flink flink-clients_2.11 1.13.1 org.apache.flink flink-connector-kafka_2.11 1.13.1 org.apache.flink flink-json 1.13.1 org.apache.flink flink-jdbc_2.11 1.10.0 data-analysis-flink org.apache.maven.plugins maven-shade-plugin 3.2.4 package shade false com.google.code.findbugs:jsr305 org.slf4j:* log4j:* *:* module-info.class META-INF/*.SF META-INF/*.DSA META-INF/*.RSA META-INF/spring.handlers reference.conf META-INF/spring.factories META-INF/spring.schemas com.wfl.firefighting.flink.DataAnalysisFlinkApplication - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
说明：
1、如果使用local执行方式，不需要提交到flink服务端执行job，可以使用spring-boot-maven-plugin，直接java -jar执行即可，如下：
data-analysis-flink org.springframework.boot spring-boot-maven-plugin com.wfl.firefighting.flink.DataAnalysisFlinkApplication repackage使用spring-boot-maven-plugin打的jar包，提交到flink集群端执行，会报错，提示找不到类，因为springboot默认打包BOOT-INF目录，flink服务端执行会提示找不到类。使用maven-shade-plugin打包，既可以用java -jar执行，也可以提交到flink服务端执行。
2、maven-shade-plugin打的jar包，如果提交到服务端执行，需要去掉springboot默认集成的logback，否则服务端执行报错，提示Caused by: java.lang.IllegalArgumentException: LoggerFactory is not a Logback LoggerContext but Logback is on the classpath，如下：
org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-logging如果本地执行java -jar形式，需要在build的中注释掉以下内容，否则启动报错提示：java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory
3、使用maven-shade-plugin打包，必须添加如下，否则提示Cannot find ‘resource’ in class org.apache.maven.plugins.shade.resource.ManifestResourceTransformer
META-INF/spring.handlers reference.conf META-INF/spring.factories META-INF/spring.schemas com.wfl.firefighting.flink.DataAnalysisFlinkApplication七、执行效果：
1、本地执行
2、提交到flink集群执行
八、其他踩坑点
1、报错提示：The RemoteEnvironment cannot be instantiated when running in a pre-defined context
解决方法：将StreamExecutionEnvironment修改为getExecutionEnvironment，获取当前执行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();- 1
2、报错提示：Insufficient number of network buffers: required 65, but only 38 available. The total number of network buffers is currently set to 2048 of 32768 bytes each.
解决办法：env.setParallelism(1)
env.setParallelism(1);- 1
3、报错提示：Caused by: java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’
解决办法：在 flink 配置文件里 flink-conf.yaml设置
classloader.check-leaked-classloader: false先自我介绍一下，小编13年上师交大毕业，曾经在小公司待过，去过华为OPPO等大厂，18年进入阿里，直到现在。深知大多数初中级java工程师，想要升技能，往往是需要自己摸索成长或是报班学习，但对于培训机构动则近万元的学费，着实压力不小。自己不成体系的自学效率很低又漫长，而且容易碰到天花板技术停止不前。因此我收集了一份《java开发全套学习资料》送给大家，初衷也很简单，就是希望帮助到想自学又不知道该从何学起的朋友，同时减轻大家的负担。添加下方名片，即可获取全套学习资料哦
-
相关阅读:
Python基础——文件系统（os模块和os.path模块）
Node.js代码漏洞扫描工具介绍——npm audit
Redis + NodeJS 实现一个能处理海量数据的异步任务队列系统
第12章 类和动态内存分配
ffmpeg.dll在哪个文件夹？怎么快速的修复ffmpeg.dll文件的丢失
PolarDB for PostgreSQL透明加密
【Linux】基础：进程的概念
十二、流程控制-循环
【Pytorch深度学习实战】（6）递归神经网络（RNN）
ue4学习日记3
- 原文地址：https://blog.csdn.net/m0_67391518/article/details/126115663
