Kafka连接自定义转换，将无模式Json转换为Avro

Question

我正试图构建一个系统，从Kafka读取json数据(无模式(，将其转换为avro并将其推送到s3。

我已经能够使用KStreams和KSQL实现json到avro的转换。我想知道使用Kafka Connect的自定义转换是否也可以实现同样的事情。

这就是我迄今为止所尝试的：

public class JsontoAvroConverter<R extends ConnectRecord<R>> implements Transformation<R> {
public static final String OVERVIEW_DOC = "Transform Payload to Custom Format";
private static final String PURPOSE = "transforming payload";
public static final ConfigDef CONFIG_DEF = new ConfigDef();
@Override
public void configure(Map<String, ?> props) {
}
@Override
public ConfigDef config() {
return CONFIG_DEF;
}
@Override
public void close() {
}
@Override
public R apply(R record) {
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("acks", "1");
properties.setProperty("retries", "10");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", KafkaAvroSerializer.class.getName());
properties.setProperty("schema.registry.url", "http://127.0.0.1:8081");
avro_Schema updatedSchema = makeUpdatedSchema();
return newRecord(record, updatedSchema);
}
private avro_Schema makeUpdatedSchema() {
avro_Schema.Builder avro_record = avro_Schema.newBuilder()
.setName("test")
.setTry$(1);
return avro_record.build();
}
protected Object operatingValue(R record) {
return record.value();
}
protected R newRecord(R record, avro_Schema updatedSchema) {
return record.newRecord(record.topic(), record.kafkaPartition(), record.keySchema(), record.key(), updatedSchema, record.value(), record.timestamp());
}
}

其中avro_schema是在avsc文件中指定的模式的名称。

我不确定这是否是正确的方法，但我面临的问题是，当调用newRecord((函数时，它希望updatedSchema是Schema类型，但我为它提供了一个自定义的avro_Schema类型。

此外，我保存到updatedSchema中的avro_record.build((并不是真正的模式，而是转换后的记录本身。但我不能只将记录主题、键(=null(和updatedRecord传递给newRecord函数。它需要分别使用模式和值。

我的问题是：

1. 是否可以使用KafkaConnect将json转换为avro，而不使用KStreams或KSQL？-因为这两种备选方案都需要设置独立的服务
2. 如何将自定义avro模式传递给newRecord函数，然后单独提供数据

如果这个问题已经得到回答，我很抱歉，我确实问了一些其他问题，但似乎没有一个能回答我的疑问。如果你需要任何其他细节，请告诉我。非常感谢。

Answer 1

KafkaConnect自定义转换器只需要向传入的JSON添加一个模式。接收器属性format.class=io.confluent.connect.s3.format.avro.AvroFormat将处理其余部分。

如果没有架构，记录值就是Map，而有了架构，它就变成了结构。我不得不修改我的代码如下：

@Override
public R apply(R record) {
final Map<String,?> value = requireMap(record.value(),PURPOSE);
Schema updatedSchema = makeUpdatedSchema();
final Struct updatedValue = new Struct(updatedSchema);

for (Field field : updatedSchema.fields()) {
updatedValue.put(field.name(), value.get(field.name()));
}

return newRecord(record, updatedSchema, updatedValue);
}
private Schema makeUpdatedSchema() {
final SchemaBuilder builder = SchemaBuilder.struct()
.name("json_schema")
.field("name",Schema.STRING_SCHEMA)
.field("try",Schema.INT64_SCHEMA);
return builder.build();
}

感谢@OneCricketer澄清我的疑虑！