Consuming Kafka Messages From Apache Flink

In my previous post, I introduced a simple Apache Flink example, which just listens to a port and streams whatever the data posts on that port. Now, it is time to see a more realistic example.

You may also like:  Kafka Producer and Consumer Examples Using Java

In the real world, we publish our streaming data to queue-like structures; it may be a JMS queue or Kafka topic.

I will not explain how to install and run Kafka, but if you need to learn you can see it in "Apache Kafka In Action"

Now, I assume you have at least one running Kafka instance.

Our business problem is to listen to customer creation events and show the number of customers created per country.

First, we will create a stream execution environment, and create a Kafka consumer object to consume messages from Kafka.

final StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "customerAnalytics");

FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("customer.create", new SimpleStringSchema(), properties);

Here we will be listening to the "customer.create" topic.

Now, we can create a Flink DataStream on top of the Kafka consumer object:

DataStream<String> stream = see.addSource(kafkaSource);

We should convert this data stream into a keyed one:

KeyedStream<Customer, String> customerPerCountryStream = stream.map(data -> {
try {
return OM.readValue(data, Customer.class);
} catch (Exception e) {
LOG.info("exception reading data: " + data);
return null;
}
}).filter(Objects::nonNull).keyBy(Customer::getCountry);

An aggregation is to be created on the stream to have some statistics. We aim to get the number of customer creation per country, so we are to create an aggregator.

public class CustomerAggregatorByCountry
implements AggregateFunction<Customer, Tuple2<String, Long>, Tuple2<String, Long>> {

/**
 * 
 */
private static final long serialVersionUID = -8528772774907786176L;

private static final Logger LOG = LoggerFactory.getLogger(CustomerAggregatorByCountry.class);

@Override
public Tuple2<String, Long> createAccumulator() {
return new Tuple2<String, Long>("", 0L);
}

@Override
public Tuple2<String, Long> add(Customer value, Tuple2<String, Long> accumulator) {
accumulator.f0 = value.getCountry();
accumulator.f1 += 1;
return accumulator;
}

@Override
public Tuple2<String, Long> getResult(Tuple2<String, Long> accumulator) {
return accumulator;
}

@Override
public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
return new Tuple2<String, Long>(a.f0, a.f1 + b.f1);
}

}

The CustomerAggregatorByCountry class implements four methods; the createAccumulator  method runs when a new key instance is encountered, the add method runs when a new instance of an existing key is encountered, and the merge method runs incase of parallel execution and when two accumulators emerge and needed to be merged. Finally, the getResult method runs when the accumulator is requested by the client application.

Now, we can apply the aggregation to our datastream as in the following code:

DataStream<Tuple2<String, Long>> result = customerPerCountryStream
    .timeWindow(Time.seconds(5))
.aggregate(new CustomerAggregatorByCountry());

We have a 5 seconds time window for our aggregation. We can also use a countWindow.

Finally, it is time to see the result. We can persist it to a database or another Kafka topic, but for the sake of simplicity, I will just print it into the console.

result.print();
see.execute("CustomerRegistrationApp");

In order to test the application, you can feed the following data to Kafka via kafka_console_producer.

$ ./kafka-console-producer.sh --broker-list localhost:9092 --topic customer.create
> {"id":1, "first":"Dursun", "last":"KOC", "country":"JP"}
> {"id":2, "first":"Mustafa", "last":"KOC", "country":"GB"}
> {"id":3, "first":"Yasemin", "last":"KOC", "country":"TR"}
> {"id":4, "first":"Ihsan", "last":"KOC", "country":"TR"}
> {"id":5, "first":"Neziha", "last":"KOC", "country":"TR"}
> {"id":6, "first":"Elif Nisa", "last":"KOC", "country":"TR"}
> {"id":7, "first":"Beyza", "last":"KOC", "country":"TR"}
> {"id":8, "first":"Zeynep", "last":"KOC", "country":"TR"}
> {"id":9, "first":"Murat", "last":"KOC", "country":"USA"}
> {"id":10, "first":"Temur", "last":"KOC", "country":"USA"}
> {"id":11, "first":"Hakan", "last":"KOC", "country":"JP"}
> {"id":12, "first":"Cemil", "last":"KOC", "country":"GB"}
> {"id":13, "first":"Turan", "last":"KOC", "country":"TR"}
> {"id":14, "first":"Hamide", "last":"KOC", "country":"TR"}
> {"id":15, "first":"Hayrettin", "last":"KOC", "country":"TR"}
> {"id":16, "first":"Fuat", "last":"KOC", "country":"TR"}
> {"id":17, "first":"Rasim", "last":"KOC", "country":"TR"}
> {"id":18, "first":"Ali Ihsan", "last":"KOC", "country":"TR"}
> {"id":19, "first":"Ali Osman", "last":"KOC", "country":"TR"}
> {"id":20, "first":"Hamit", "last":"KOC", "country":"USA"}

You can find the full running application at my GitHub repository: https://github.com/dursunkoc/flink-kafka-sample. If you have any questions about the implementation, you can comment below.

Further Reading

A Tutorial on Kafka With Spring Boot