flink 缓存问题处理过程，看我十八般武艺

Posted 2023-03-02 香菜+

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了flink 缓存问题处理过程，看我十八般武艺相关的知识，希望对你有一定的参考价值。

1、问题描述

作为flink小白，第一次写flink的任务，所以战战兢兢的复写了官方的demo，还好一切顺利，自信心倍增，开始进入业务需求。

但是作为一个新手不踩几个坑那说不过去，所以遇上了第一个大的门槛。

1.1 版本信息

flink版本：1.15.2

java 版本：jdk11

OS：Ubuntu 18.04

1.2 lib 包冲突的问题

第一次使用flink 哪知道flink还有个lib 目录，在使用的过程中出现了包的冲突，主要其他同学在测试的过程中瞎鸡儿搞，放了几个包在lib下，也莫名其妙，不过好在身边有人帮忙解决了问题，将maven中的flink提供的核心包设置为provided，这个很简单。

1.3 缓存问题

这个问题最初不是我遇到的，是同事遇到了，但是花了好多时间也没解决，就放弃了。我接过来处理。这里说下背景，flink我们使用的是standalone 模式，并且是另外一个同事安装的。也不知道做了什么配置，也是第一次写flink，甚至都不知道怎么提交任务，在同事手把手教的情况下上车，所以直接上手难度挺高。

开发的一个任务，提交之后，因为有问题，所以直接报错了，看下报错日志。

看日志还是挺明显的，proto解析失败，遇到bug解决bug，所以后面修改代码，做了异常捕获，但是不管怎么修改，每次提交都是包这个错误，真的是头秃，解决不了。

看下等会一直在说的代码，这个类主要是做kafka数据的解析，将kafka中的数据解析成VehData。

public class BsmSchema implements DeserializationSchema<BsmDTO.VehData> 
    @Override
    public BsmDTO.VehData deserialize(byte[] bytes) throws IOException 
        try 
            if (Objects.nonNull(bytes) && bytes.length != 0) 
                BsmDTO.VehData bsm = BsmDTO.VehData.parseFrom(bytes);
                return bsm;
            
         catch (Exception e) 
        
        return null;
    
    @Override
    public boolean isEndOfStream(BsmDTO.VehData dpe) 
        return false;
    
    @Override
    public TypeInformation<BsmDTO.VehData> getProducedType() 
        return TypeInformation.of(new TypeHint<>() 
        );

2、解决过程

2.1 proto 解析失败

这个还是很简单的，因为数据的问题导致数据解析失败，所以直接通过try catch 对异常数据不处理，在本地测试通过后是正常的，打包之后，上传jar ，直接上面同样的错误。无语

2.2 确认try catch

因为报了同样的错误，所以认为没有打包进最新的代码，直接反编译代码，确认代码中存在try catch

2.3 去除proto解析的问题

因为一直报proto解析错误，那我就索性去掉了proto的解析部分，我直接返回了一个null，可以看到不会再有proto解析了。

    public BsmDTO.VehData deserialize(byte[] bytes) throws IOException 
//        try 
//            if (Objects.nonNull(bytes) && bytes.length != 0) 
//                BsmDTO.VehData bsm = BsmDTO.VehData.parseFrom(bytes);
//                return bsm;
//            
//        catch (Exception e) 
//        
        return null;

结果就是同样的报错，BsmSchema 的15行，这就很奇怪了，代码里根本没这一行代码，当然也通过反编译确认了没有这个代码。这里基本上验证了Flink运行的根本不是我刚刚提交的最新代码。真的是奇怪啊，这里我也确认了缓存的存在。