使用kafka-connect-transform-archive和HdfsSinkConnector时的刷新大小

Question

我有一个Kafka主题的数据，我想把它保存在我的数据湖里。

在担心密钥之前，我能够使用HDFSSINKConnector将Avro值保存在datalake上的文件中。每个文件中的消息值的数目由HDFSSINKConnector的“Flush.Size”属性确定。

都很好。接下来，我还想保留钥匙。为此，我使用了kafka-connect-transform-archive，它将字符串、键和Avro值包装到一个新的Avro模式中。

这两种情况的唯一区别是HdfsSinkConnector的配置，它指定了存档转换。

"transforms": "tran",
"transforms.tran.type": "com.github.jcustenborder.kafka.connect.archive.Archive"

kafka-connect-transform-archive在设计上是否忽略了刷新大小，或者是否需要一些额外的配置，以便能够在数据湖上保存每个文件的多个键值消息？

Answer 1

我在使用kafka gcs接收器连接器时也遇到了同样的问题。

在com.github.jcustenborder.kafka.connect.archive.archive代码中，每个消息创建一个新的模式。

private R applyWithSchema(R r) {
final Schema schema = SchemaBuilder.struct()
    .name("com.github.jcustenborder.kafka.connect.archive.Storage")
    .field("key", r.keySchema())
    .field("value", r.valueSchema())
    .field("topic", Schema.STRING_SCHEMA)
    .field("timestamp", Schema.INT64_SCHEMA);
Struct value = new Struct(schema)
    .put("key", r.key())
    .put("value", r.value())
    .put("topic", r.topic())
    .put("timestamp", r.timestamp());
return r.newRecord(r.topic(), r.kafkaPartition(), null, null, schema, value, r.timestamp());

}