Apache Kafka is a popular distributed streaming platform that enables building real-time data pipelines and streaming applications. Kafka Streams is a client library for building applications and microservices, where the input and output data are stored in Kafka clusters. In this discussion, we focus on a common data transformation scenario: streaming data from JSON format to Avro format using Kafka Streams.

Key Concepts and Technologies

JSON (JavaScript Object Notation) is a lightweight data interchange format that is easy for humans to read and write and easy for machines to parse and generate. It is commonly used in various data exchange scenarios in web applications.

Avro is a data serialization system that enables efficient and compact binary format data exchange. It is often used in Apache Kafka to ensure schema management and compatibility across distributed data streams.

Kafka Streams is the stream processing library provided by Apache Kafka. It allows for stateful and stateless processing, windowing, and complex event processing by transforming input Kafka topics into output Kafka topics.

Schema Registry is a service provided by Confluent (and compatible with other platforms) that manages Avro schemas and their versions, ensuring that only valid Avro data is published to Kafka topics.

Processing Flow: From JSON to Avro

The typical processing flow involves reading messages from a Kafka topic with data in JSON format, transforming these messages into Avro format, and then writing the transformed messages back to a different Kafka topic. To accomplish this, developers need to use Kafka Streams for the transformation process, which can be broken down into a few steps:

1. Read from Kafka: Stream messages from a Kafka topic with JSON values.
2. Deserialize JSON: Convert JSON string messages into Java objects.
3. Serialize to Avro: Convert Java objects into Avro format.
4. Write to Kafka: Stream the Avro messages back to a Kafka topic.

Example Workflow with Code Snippets

Assuming you have Kafka and Schema Registry running, you can implement the Kafka Streams application using Java:

1import org.apache.kafka.common.serialization.Serdes;
2import org.apache.kafka.streams.KafkaStreams;
3import org.apache.kafka.streams.StreamsBuilder;
4import org.apache.kafka.streams.kstream.KStream;
5import org.apache.kafka.streams.kstream.ValueMapper;
6import org.apache.kafka.streams.StreamsConfig;
7import com.fasterxml.jackson.databind.JsonNode;
8import com.fasterxml.jackson.databind.ObjectMapper;
9import org.apache.avro.generic.GenericRecord;
10import io.confluent.kafka.serializers.KafkaAvroSerializer;
11import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
12
13import java.util.Properties;
14
15public class JsonToAvroStreamProcessor {
16    public static void main(String[] args) {
17        Properties props = new Properties();
18        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "json-to-avro-stream-app");
19        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
20        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
21        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
22
23        StreamsBuilder builder = new StreamsBuilder();
24
25        // read from JSON topic
26        KStream<String, String> sourceStream = builder.stream("json-input-topic");
27
28        // transform to Avro
29        KStream<String, GenericRecord> avroStream = sourceStream.mapValues((ValueMapper<String, GenericRecord>) JsonToAvroStreamProcessor::convertJsonToAvro);
30
31        // write to Avro topic
32        avroStream.to("avro-output-topic", Produced.with(Serdes.String(), new GenericAvroSerde()));
33
34        KafkaStreams streams = new KafkaStreams(builder.build(), props);
35        streams.start();
36
37        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
38    }
39
40    private static GenericRecord convertJsonToAvro(String json) {
41        // Assume schema and object mapping logic exists
42        // Implement JSON to Avro conversion here
43    }
44}

Table of Key Points

Additional Considerations

• Schema Evolution: Avro supports schema evolution, allowing you to modify schemas without breaking existing applications. Schema Registry helps manage this by enforcing compatibility rules.
• Performance: Avro's binary format generally offers better performance and compression compared to JSON. This is crucial in high-throughput environments such as Kafka.
• Tooling and Ecosystem: Avro is deeply integrated with the Kafka ecosystem, and tools like Schema Registry provide essential services that enhance the robustness of Kafka's data handling capabilities.

Conclusion

Converting JSON to Avro within a Kafka Streams application involves deserializing JSON into Java objects, transforming these objects according to an Avro schema, and serializing them into Avro format. By leveraging Kafka Streams along with Schema Registry, developers can build robust and efficient streaming applications that capitalize on Kafka's powerful data streaming capabilities.