使用kafka流的Topic1到Topic2

Question

我是一个新的Kafka流，我想阅读一个主题，并在一个新的主题中使用Kafka流API写它的一部分。我的键是string，值是Avro有文档/示例可以使用吗？

编辑：

    final StreamsBuilder builder = new StreamsBuilder();
    final KStream<String, GenericRecord> inputStream = builder.stream("Test_CX_TEST_KAFKA_X");
    final KStream<String, String> newStream = inputStream.mapValues(value -> value.get("ID").toString());
    newStream.to("SUB_TOPIC",Produced.with(Serdes.String(),Serdes.String()));
    final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
    streams.start();

在分题中，我有：

如何在新主题中添加Body中的其他字段？示例：

{“id”：“145”,“timestamp”：1552585938545,“week”：“\u0000”,“source”：{“string”：“tmp”},“body”：{“string”：“{\”operation_type\“：\”insert\“,\”old\“：{\”row_id\“：null,\”last_upd\“：null},\”new\“：{\”row_id\“：\”170309-*********\“,\”last_upd\“：\”2019-03-14t17:52:18\“}}”},“

Answer 1

您可以简单地将主题作为流使用，并使用.map（）/.mapvalues（）函数修改Value/KeyValues。

例如：假设您想从avro记录中选择一个列并发布到新的输出主题。

// If you are using Schema registry, make sure to add the schema registry url 
// in streamConfiguration. Also specify the AvroSerde for VALUE_SERDE

final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, GenericRecord> inputStream = builder.stream("inputTopic");
final KStream<String, String> newStream = userProfiles.mapValues(value -> value.get("fieldName").toString());
subStream.to("outputTopic",Produced.with(Serdes.String(),Serdes.String());
final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);

此外，您还可以查看github上的示例:
https://github.com/confluentinc/kafka-streams-examples/blob/5.1.2-post/src/main/java/io/confluent/examples/streams/wikipedaFeedavroexample.java