通过自定义flume插件读取systcp类型数据中的protobuf格式消息

由于项目中有大量的日志数据需要收集,并且后续需要对日志进行进一步的分析,考虑使用flume进行日志采集,存储到hdfs和hbase上用于后续的分析。

flume自带的有spool、systcp、exec等source类型,刚开始调试的使用。发现日志源通过systcp方式发送到flume的时候,接受不到数据,通过排查得知,日志源产生的日志格式是protobuf格式的。

为了减少日志生产者的响应时间,不能让日志生产者把protobuf格式转换为普通文本方式传输,同时使用protobuf的目的就是跨语言的序列化,和json相比具有文件小等优点,所以就考虑自定义flume的source类型,在得到tcp数据后通过protobuf的解码器对得到的byte数组就行解码,然后封装成flume需要的Event格式。

在此过程中犯了一个很严重的问题,因为日志的产生是由另外一个同事实现的,是用c++封装的protobuf格式,让他提供了一个产生测试数据的测试程序,但是接收到的数据,用protobuf解码器解码的时候,总是报错,说是包含不合法的。不管怎么测试都是报错,我就一直在查看flume中使用的netty获取tcp数据是否不完整的原因,但是一直没有找到netty是如何对获取的数据分段的设置。按说netty获取tcp socket数据应该就是客户端发送一次就会接受一次,不会出现自动截断的问题。

后来我从那个同事拿到他用protobuf编码之前的数据,我使用java的protobuf编码器编码成二进制文件后,在服务端解析是能够成功的,所以就怀疑那个同事提供的产生日志的测试程序有问题,后来让他检查后发现确实是多发了数据,导致数据包大小不一致。

从这次事件得到的教训就是在和其他人进行数据交互的时候,不能直接就从别人那里拿数据来验证,首先要保证自己从原始数据编解码能够成功后在和别人进行交互,就不会浪费时间了。

主要的原因还是我自己对服务端的开发不熟悉,对于byte数据的操作和socket传输数据也不熟悉的缘故吧,进一步学习吧。


EOF