https://danlebrero.com/2017/01/05/proof-of-concept-using-kafkastreams-and-ktables/
In the last article, we used kafka consumer and producer to read in and write out the kafka messages. In this article, we will show how to replace them with Kafka Streams so that we don’t have to handle back pressure ourselves. Back-pressure is an important feedback mechanism that allows systems to gracefully respond to load rather than collapse under it.
Kafka Streams is a new open source library, part of the Apache Kafka project, that promises to make stream processing simple without losing the power and scalability of other stream processing systems like Storm or Spark Streaming
The major selling points are:
- Scalable, using the same partition-based model as Kafka.
- Real time, optional micro-batching.
- Stateful stream processing.
- It is a library, not a framework.
Kafka Streams don’t need any new infrastructure, depending only on the Kafka cluster. It has a nice functional API similar to Java 8 reactive streams.
Configure Kafka Streams
To start Kafka streams in your Java project, you’d need to define the following properties:
Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, kafkaStreamConfig.getApplicationID()); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaStreamConfig.getBootstrapServer()); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); StreamsBuilder builder = new StreamsBuilder(); KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start();
You’d also need to specify the POJO deserializer forKafka stream when reading in the messages as dataPointSerde:
Map<String, Object> serdeProps = new HashMap<>();
final Serializer<DataPoint> dataPointSerializer = new JsonPOJOSerializer<>();
serdeProps.put("JsonPOJOClass", DataPoint.class);
dataPointSerializer.configure(serdeProps, false);
final Deserializer<DataPoint> dataPointDeserializer = new JsonPOJODeserializer<>();
serdeProps.put("JsonPOJOClass", DataPoint.class);
dataPointDeserializer.configure(serdeProps, false);
final Serde<DataPoint> dataPointSerde = Serdes.serdeFrom(dataPointSerializer, dataPointDeserializer);
Introducing Java Reactive Streams
http://www.reactive-streams.org/
Java Reactive Streams is useful for handling streams of data—especially “live” data whose volume is not predetermined—requires special care in an asynchronous system. The most prominent issue is that resource consumption needs to be controlled such that a fast data source does not overwhelm the stream destination. Asynchrony is needed in order to enable the parallel use of computing resources, on collaborating network hosts or multiple CPU cores within a single machine (see more on Reactive Manifesto). The scope of Reactive Streams is to find a minimal set of interfaces, methods and protocols that will describe the necessary operations and entities to achieve the goal—asynchronous streams of data with non-blocking back pressure.
First, to transform the input Kafka Stream message, we can the KStream interface ValueMapper and define the logic for transform input DataPoint to LabeledDataPoint in the apply method.
public class LabeledDataPointProducer implements ValueMapper<DataPoint, LabeledDataPoint> {
private final ShotoClassifier classifier;
private final DataPointCassandraDao cassandraDao;
public LabeledDataPointProducer(ShotoClassifier classifier, DataPointCassandraDao cassandraDao) {
this.classifier = classifier;
this.cassandraDao = cassandraDao;
}
/**
* Obtains a label for the datapoint provided either through cache or classification
* @param dataPoint
* @return a LabeledDataPoint
*/
public LabeledDataPoint apply(DataPoint dataPoint) {
LabeledDataPoint cached = cassandraDao.getCached(dataPoint.getMd5String());
// if cached then set isCached to true and return the labeledDataPoint
if (cached != null) {
cached.setCached(true);
return cached;
}
// if not cached then transform and save the labeledDataPoint to cassandra database
LabeledScore<CategoryClassification> label = classifier.classify(dataPoint.getInputText()).getLabel();
LabeledDataPoint ldp = LabeledDataPoint.fromClassification(dataPoint, label);
cassandraDao.saveDataPoint(ldp, ldp.isHuman());
return ldp;
}
}
We can implement Consumer interface (consume input, return void) that takes the type of kafka message “LabeledDataPoint" and pass into the checking logic defined in the override accept method:
public class OutboundConsumer implements Consumer<LabeledDataPoint> {
final Random random = new Random(1L);
private double confidenceThreshold;
private double highConfAuditPct;
private String highConfTag;
private String lowConfTag;
private BiConsumer<LabeledDataPoint, String> writeToOutbound;
public OutboundConsumer (double confidenceThreshold, double highConfAuditPct, String highConfTag, String lowConfTag, BiConsumer<LabeledDataPoint, String>writeToOutbound) {
this.confidenceThreshold = confidenceThreshold;
this.highConfAuditPct = highConfAuditPct;
this.highConfTag = highConfTag;
this.lowConfTag = lowConfTag;
this.writeToOutbound =writeToOutbound;
};
@Override
public void accept(LabeledDataPoint labeledDataPoint) {
if (labeledDataPoint.isCached()) {
return;
} else if (labeledDataPoint.getLabelConfidence() < confidenceThreshold) {
writeToOutBound.accept(labeledDataPoint, lowConfTag);
} else if (random.nextFloat() < highConfAuditPct) {writeToOutBound.accept(labeledDataPoint, highConfTag);
}
}
@Override
public Consumer<LabeledDataPoint> andThen(Consumer<? super LabeledDataPoint> after) {
return null;
}
}
After preparing all the above, we can now pass in Java Reactive Streams to in Kafka Stream API to transform our input data as we intended with:
// read in and transform input DataPoint to LabeledDataPoint with ValueMapper
KStream<String, LabeledDataPoint> dataPointKStream = builder.stream(kafkaStreamConfig.getStreamInboundTopic(),
Consumed.with(Serdes.String(), dataPointSerde)).mapValues(labeledDataPointProducer);
// check condition and write result to kafka streams
dataPointKStream.filter((k,v) -> v.isCached()).filter((k,v) -> v.getLabelConfidence() > 0.1)
.to(kafkaStreamConfig.getStreamOutboundTopic());
// pass in Java consumer
dataPointKStream.foreach((k,v) -> outboundConsumer.accept(v));
