我正在寻找kafka connect hdfs connector(confluent)支持,以使用flatbuffer模式保存字节数组和字段分区。
我正在从Kafka接收字节数组中的数据。这个字节数组是从flatbuffer生成的。需要将其保存在hdfs中,路径为field1/field2/field3。所有这些字段都需要使用flatbuffer模式从字节数组中提取。另外,要保存在hdfs中的数据只需要以字节为单位。数据不需要转换。
我都查过了:
字段分隔符:https://github.com/confluentinc/kafka-connect-storage-common/blob/master/partitioner/src/main/java/io/confluent/connect/storage/partitioner/fieldpartitioner.java
支持的格式:json、avro、parquet。在https://github.com/confluentinc/kafka-connect-storage-cloud/blob/master/kafka-connect-s3/src/main/java/io/confluent/connect/s3/format/json/jsonrecordwriterprovider.java,但如果数据类型为kafka struct,我发现bytearray保存在hdfs中。
我找不到一个方法来利用他们为我的目的。
有人知道这种内置支持吗。如果没有,那么请指导我的资源(如果有的话)建立自定义支持两者。
1条答案
按热度按时间g0czyy6m1#
flatbuffers(当前)不是受支持的序列化格式,并且
ByteArrayFormat
仅适用于s3 connect,不适用于hdfs,并且只转储ByteArraySerializer
来自kafka的格式(在转换器之后是一个struct对象,是的。至于分区,由于数据只有字节,它不检查记录值以支持分区器,因此您还需要添加一个自定义值,这将需要对消息进行反序列化以检查字段。
我不知道为什么要链接到s3connect代码,但是如果要添加自己的格式,请查看将stringformat添加到hdfs connect的pr
要构建项目,请查看常见问题解答