Flink 开发生产问题汇总,亲自解决的才是最宝贵的

2024-05-02 07:38

本文主要是介绍Flink 开发生产问题汇总,亲自解决的才是最宝贵的,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

本文收集了与粉丝沟通过程中常见的问题与解决方案,整理成文,供大家参考和查阅。

1、Checkpoint失败:Checkpoint expired before completing

原因是因为

checkpointConf.setCheckpointTimeout(8000L)

设置的太小了,默认是10min,这里只设置了8sec。

当一个Flink App背压的时候(例如由外部组件异常引起),Barrier会流动的非常缓慢,导致Checkpoint时长飙升。

2、资源隔离建议

在Flink中,资源的隔离是通过Slot进行的,也就是说多个Slot会运行在同一个JVM中,这种隔离很弱,尤其对于生产环境。Flink App上线之前要在一个单独的Flink集群上进行测试,否则一个不稳定、存在问题的Flink App上线,很可能影响整个Flink集群上的App。

3、资源不足导致 container 被 kill
`The assigned slot container_container编号 was removed.`

Flink App 抛出此类异常,通过查看日志,一般就是某一个 Flink App 内存占用大,导致 TaskManager(在 Yarn 上就是 Container )被Kill 掉。

但是并不是所有的情况都是这个原因,还需要进一步看 yarn 的日志( 查看 yarn 任务日志:yarn logs -applicationId  -appOwner),如果代码写的没问题,就确实是资源不够了,其实 1G Slot 跑多个Task( Slot Group Share )其实挺容易出现的。

因此有两种选择,可以根据具体情况,权衡选择一个。

  • 将该 Flink App 调度在 Per Slot 内存更大的集群上。

  • 通过 slotSharingGroup("xxx") ,减少 Slot 中共享 Task 的个数

4、启动报错,提示找不到 jersey 的类
java.lang.NoClassDefFoundError: com/sun/jersey/core/util/FeaturesAndProperties

解决办法进入 yarn中 把 lib 目中的一下两个问价拷贝到 flink 的 lib 中

hadoop/share/hadoop/yarn/lib/jersey-client-1.9.jar /hadoop/share/hadoop/yarn/lib/jersey-core-1.9.jar

5、Scala版本冲突
java.lang.NoSuchMethodError:scala.collection.immutable.HashSet$.empty()Lscala/collection/

解决办法,添加

import org.apache.flink.api.scala._
6、没有使用回撤流报错
Table is not an append一only table. Use the toRetractStream() in order to handle add and retract messages.

这个是因为动态表不是 append-only 模式的,需要用 toRetractStream ( 回撤流) 处理就好了.

tableEnv.toRetractStream[Person](result).print()
7、OOM 问题解决思路
java.lang.OutOfMemoryError: GC overhead limit exceeded
java.lang.OutOfMemoryError: GC overhead limit exceededat java.util.Arrays.copyOfRange(Arrays.java:3664)at java.lang.String.<init>(String.java:207)at com.esotericsoftware.kryo.io.Input.readString(Input.java:466)at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.read(DefaultSerializers.java:177)
......at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:524)

解决方案:

  1. 检查 slot 槽位够不够或者 slot 分配的数量有没有生效

  2. 程序起的并行是否都正常分配了(会有这样的情况出现,假如 5 个并行,但是只有 2 个在几点上生效了,另外 3 个没有数据流动)

  3. 检查flink程序有没有数据倾斜,可以通过 flink 的 ui 界面查看每个分区子节点处理的数据量

8、解析返回值类型失败报错
The return type of function could not be determined automatically
Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'main(RemoteEnvironmentTest.java:27)' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.at org.apache.flink.api.java.DataSet.getType(DataSet.java:178)at org.apache.flink.api.java.DataSet.collect(DataSet.java:410)at org.apache.flink.api.java.DataSet.print(DataSet.java:1652)

解决方案:产生这种现象的原因一般是使用 lambda 表达式没有明确返回值类型,或者使用特使的数据结构 flink 无法解析其类型,这时候我们需要在方法的后面添加返回值类型,比如字符串

input.flatMap((Integer number, Collector<String> out) -> {......
})
// 提供返回值类型
.returns(Types.STRING)
9、Hadoop jar 包冲突
Caused by: java.io.IOException: The given file system URI (hdfs:///data/checkpoint-data/abtest) did not describe the authority (like for example HDFS NameNode address/port or S3 host). The attempt to use a configured default authority failed: Hadoop configuration did not contain an entry for the default file system ('fs.defaultFS').at org.apache.flink.runtime.fs.hdfs.HadoopFsFactory.create(HadoopFsFactory.java:135)at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:399)at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:318)at org.apache.flink.core.fs.Path.getFileSystem(Path.java:298)

解决:pom 文件中去掉和 hadoop 相关的依赖就好了

10、时钟不同步导致无法启动

启动Flink任务的时候报错

Caused by: java.lang.RuntimeException: Couldn't deploy Yarn cluster

然后仔细看发现里面有这么一句

system times on machines may be out of sync

意思说是机器上的系统时间可能不同步。同步集群机器时间即可。

这篇关于Flink 开发生产问题汇总,亲自解决的才是最宝贵的的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/953732

相关文章

SpringBoot内嵌Tomcat临时目录问题及解决

《SpringBoot内嵌Tomcat临时目录问题及解决》:本文主要介绍SpringBoot内嵌Tomcat临时目录问题及解决,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录SprinjavascriptgBoot内嵌Tomcat临时目录问题1.背景2.方案3.代码中配置t

SpringBoot使用GZIP压缩反回数据问题

《SpringBoot使用GZIP压缩反回数据问题》:本文主要介绍SpringBoot使用GZIP压缩反回数据问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录SpringBoot使用GZIP压缩反回数据1、初识gzip2、gzip是什么,可以干什么?3、Spr

如何解决idea的Module:‘:app‘platform‘android-32‘not found.问题

《如何解决idea的Module:‘:app‘platform‘android-32‘notfound.问题》:本文主要介绍如何解决idea的Module:‘:app‘platform‘andr... 目录idea的Module:‘:app‘pwww.chinasem.cnlatform‘android-32

kali linux 无法登录root的问题及解决方法

《kalilinux无法登录root的问题及解决方法》:本文主要介绍kalilinux无法登录root的问题及解决方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,... 目录kali linux 无法登录root1、问题描述1.1、本地登录root1.2、ssh远程登录root2、

使用Python开发一个带EPUB转换功能的Markdown编辑器

《使用Python开发一个带EPUB转换功能的Markdown编辑器》Markdown因其简单易用和强大的格式支持,成为了写作者、开发者及内容创作者的首选格式,本文将通过Python开发一个Markd... 目录应用概览代码结构与核心组件1. 初始化与布局 (__init__)2. 工具栏 (setup_t

SpringBoot应用中出现的Full GC问题的场景与解决

《SpringBoot应用中出现的FullGC问题的场景与解决》这篇文章主要为大家详细介绍了SpringBoot应用中出现的FullGC问题的场景与解决方法,文中的示例代码讲解详细,感兴趣的小伙伴可... 目录Full GC的原理与触发条件原理触发条件对Spring Boot应用的影响示例代码优化建议结论F

MySQL 中查询 VARCHAR 类型 JSON 数据的问题记录

《MySQL中查询VARCHAR类型JSON数据的问题记录》在数据库设计中,有时我们会将JSON数据存储在VARCHAR或TEXT类型字段中,本文将详细介绍如何在MySQL中有效查询存储为V... 目录一、问题背景二、mysql jsON 函数2.1 常用 JSON 函数三、查询示例3.1 基本查询3.2

Spring Shell 命令行实现交互式Shell应用开发

《SpringShell命令行实现交互式Shell应用开发》本文主要介绍了SpringShell命令行实现交互式Shell应用开发,能够帮助开发者快速构建功能丰富的命令行应用程序,具有一定的参考价... 目录引言一、Spring Shell概述二、创建命令类三、命令参数处理四、命令分组与帮助系统五、自定义S

Pyserial设置缓冲区大小失败的问题解决

《Pyserial设置缓冲区大小失败的问题解决》本文主要介绍了Pyserial设置缓冲区大小失败的问题解决,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面... 目录问题描述原因分析解决方案问题描述使用set_buffer_size()设置缓冲区大小后,buf

PyInstaller打包selenium-wire过程中常见问题和解决指南

《PyInstaller打包selenium-wire过程中常见问题和解决指南》常用的打包工具PyInstaller能将Python项目打包成单个可执行文件,但也会因为兼容性问题和路径管理而出现各种运... 目录前言1. 背景2. 可能遇到的问题概述3. PyInstaller 打包步骤及参数配置4. 依赖