Flink业务常见故障多案例解决方案

#Flink常见故障 #大数据 #生产环境真实案例 #Flink #流式计算 #流批一体 #整理 #经验总结
说明：此篇总结Flink常见故障案例处理方案 联合自身经历 总结不易 +关注 +收藏 欢迎留言
专栏：Flink业务常见故障多案例解决方案 【1】任务运行常见故障
目录
【1】任务运行常见故障
Flink对接HBase任务运行一段时间后忽然中断
Flink对接kafka-写入数据倾斜，部分分区没有写入数据
Flink任务的日志目录增长太快，导致磁盘写满
Flink消费kafka任务停止一段时间后，任务无法重启
业务运行一段时间报InvalidAMRMToken from xxx
任务启动失败，报资源不足：Could not allocate all requires slots within timeout of xxx ms
算子的部分节点产生背压，其它节点正常
FlinkSQL获取数据格式为csv、json时遇到错误数据导致任务失败，报错Caused by：org.apache.flink.api.common.io.parseException: Row too short
Flink对接HBase、ES性能弱于kafka的组件时出现持续背压而且出现Full GC
Flink写kafka出现报错Failed to send data to Kafka: This server is not the leader for that topic-partition
Flink任务运行过程中出现报错netty.exception.RemoteTransportException
Flink任务运行过程中出现报错netty.exception.LocalTransportException
Flink任务运行过程sink端中对接kafka报错
Zookeeper leader倒换后Flink任务失败
Flink任务的taskmanager.out文件太大导致磁盘使用率过高
Flink提交任务下令实行错误NoSuchMethodError
Flink客户端实行flink下令报错ClusterRetrieveException
yarn-session创建Flink集群报错NoAuthException: KeeperErrorCode = NoAuth for /flink/application_xx
Flink提交任务下令实行错误ClassNotFoundException

---

【1】任务运行常见故障

Flink对接HBase任务运行一段时间后忽然中断

问题
Flink对接HBase， 观察Flink的运行日志发现，任务运行一段时间后持续打印以下日志：
RPC Server Kerberos principal name for service=ClientService is hbase/xxxx.xxxx.xxx.com
并没有出现其它错误。
缘故起因
客户在对接HBase过程中未在代码内部定期更新单子信息，导致任务会运行一段时间后等待登录
解决方法
需要在flink-conf.ymal设置文件中调整参数：

• 打开 flink-conf.ymal设置文件，并找到设置项：env.java.opts
  
  
• 在设置项中添加-Djavax.security.auth.useSubjectCredsOnly=true
  

Flink对接kafka-写入数据倾斜，部分分区没有写入数据

问题
使用FlinkKafkaProducer进行数据生产，数据只写到了kafka的部分分区中，其它的分区没有数据写入
缘故起因

• 大概缘故起因1：Flink写kafka使用的机制与原生接口的写入方式是有差别的，在默认环境下，Flink使用了”并行度编号+分区数量”取模计算的效果作为topic的分区编号。那么会有以下两种场景：
  

• 并行度%分区数量=0，表现并行度是kafkatopic分区数的一倍或者多倍，数据的写入每个分区数据量是平衡的。
  
  
• 并行度%分区数量≠0，那么数据量势必会在个别分区上的数据量产生倾斜。
  

• 大概缘故起因2：在业务代码的部分算子中使用了keyby()方法，由于现网中的数据流中，每个key值所属的数据量不一致（就是说某些key的数据量会非常大，有些又非常小）导致每个并行度中输出的数据流量不一致。从而出现数据倾斜。
  

解决方法
缘故起因一：
方法1，调整kafka的分区数跟flink的并行度保持一致，即要求kafka的分区数与flink写kafka的sink并行度保持强一致性。这种做法的优势在于每个并行度仅需要跟每个kafka分区所在的 broker保持一个常链接即可。可以或许节省每个并发线程与分区之间调理的时间。
方法2，flink写kafka的sink的分区计谋写成随机写入模式，如下图，这样数据会随即写入topic的分区中，但是会有一部分时间损耗在线程向寻址，保举使用方法1。

缘故起因二：
需要调整业务侧对key值的选取，比方：可以将key调整为“key+随机数”的方式，包管Flink的keyby()算子中每个处理并行度中的数据是平衡的。
Flink任务的日志目录增长太快，导致磁盘写满

问题
集群告警磁盘使用率超过阈值，经过排查发现是taskmanager.out文件过大导致
缘故起因
代码中存在大量的print模块，导致taskmanager.out文件被写入大量的日志信息，taskmanager.out 一样平常是，业务代码加入了 .print的代码，需要在代码中排查是否有类似于以下的代码逻辑：

或者类似于这样的打印：

如果包罗，日志信息会持续打印到taskmanager.out里面。
解决方法
将上图红框中的代码去掉，或者输出到日志文件中。
Flink消费kafka任务停止一段时间后，任务无法重启

问题
Flink对接kafka运行一段时间手动重启任务后，任务无法重启报“Ask timed out on xxxx after [30000 ms]”。
缘故起因

• akka相干的超时时间太短，网络状态不太好。
  
  
• 内存设置太小，kafka消费的数据量太大导致任务失败。
  

解决方法

• 缘故起因一解决方案
  ：调整以下超时参数，增加akka的超时时间
  akka.ask.timeout 60s
  akka.watch.heartbeat.pause 120s
  akka.tcp.timeout 60s
  

• 缘故起因二解决方案：
  
     
  
  • 检察后台checkpoint的历史记载，是否有长时间的checkpoint超时失败。
    
    
    
    
       
    
  • 
    如果有长时间的超时失败说明，在消费kafka数据的过程中，消费偏移量没有精确提交，下次消费会从头开始消费。   
    
  • 如果数据从头开始消费，消费者初始的积压量大概会变大，通过下令./kafka-consumer-groups.sh检察积压量。   
    
  • 
       
    
  • 也可以通过kafka-consumer-group.sh下令调整消费偏移量
    （留意：该操作会导致一部分数据消费不到）
    下令如下：
    ./kafka-consumer-groups.sh --reset-offsets --to-earliest --group test --bootstrap-server kafka业务**:21007 --command-config ../config/consumer.properties --all-topics --execute**
    参数说明：
    --reset-offsets：重置offset操作。
    --to-earliest：全部分区重置到分区的第一条信息。
    --group：需要重置的groupid。
    --all-topics：表现要重置组内的全部topic，如果重置一个topic，可以写–-topic topicName。
    --execute：表现立即实行。
       
    
  • 如果积压量比较大，kafka一次消费的数据量会非常大，如果后端对接的组件性能不足（比方：ES，GUASSDB)大概会产生背压，如果taskmanager的内存不足大概会导致taskmanager出现FullGC，从而导致业务失败。
    
         
    
    • 需要在启动下令中调整taskmanager的启动内存，比方，如下下令中修改的-ytm参数
      ./flink run -m yarn-cluster -p 2 -yjm 1024 -ytm 2048 -c ……
         
      
       
    
  
  

业务运行一段时间报InvalidAMRMToken from xxx

问题
业务运行一段时间后出现InvalidAMRMToken from xxx&#x

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。