Stream Processing With Spring, Kafka, Spark and Cassandra - Part 5


This blog entry is part of a series called Stream Processing With Spring, Kafka, Spark and Cassandra.

  1. Part 1 - Overview
  2. Part 2 - Setting up Kafka
  3. Part 3 - Writing a Spring Boot Kafka Producer
  4. Part 4 - Consuming Kafka data with Spark Streaming and Output to Cassandra
  5. Part 5 - Displaying Cassandra Data With Spring Boot

Displaying Cassandra Data With Spring Boot

Now that we have our voting data in Cassandra let's write a simple Spring Boot project that simply gathers all the data from cassandra sorts them and displays to user.

Setting up a project

  1. Project SDK: Java 8
  2. Initializr Service URL: https://start.spring.io
  3. Next
  4. Name: boot-cassandra-data-show
  5. Type: Gradle Project
  6. Packaging: Jar
  7. Java Version: 1.8
  8. Language: Java
  9. Group: com.example
  10. Artifact: boot-cassandra-data-show
  11. Vesion: 0.0.1-SNAPSHOT
  12. Description: Spring Boot Display Cassandra Data
  13. Package: com.example
  14. Next
  15. Spring Boot Version: 1.3
  16. Core - Web
  17. Template Engines - Mustache
  18. Next
  19. Project name: boot-cassandra-data-show
  20. The rest is just fine ...
  21. Finish
  22. After creating project check sdk setting, it should be java 8

Cassandra dependencies


Vote class

We'll use this class to map rows from cassandra.

package com.example;

import java.io.Serializable;

public class Vote implements Serializable {
    private String name;
    private Integer votes;

    public Vote(String name, Integer votes) {
        this.name = name;
        this.votes = votes;

    public Vote() {

    public String getName() {
        return name;

    public void setName(String name) {
        this.name = name;

    public Integer getVotes() {
        return votes;

    public void setVotes(Integer votes) {
        this.votes = votes;


server.port = 8090
contactPoint =
keyspace = voting


This bean is used to setup connection towards Cassandra

package com.example;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

public class CassandraSessionManager {

    private Session session;
    private Cluster cluster;

    private String contactPoint;

    private String keyspace;

    public CassandraSessionManager() {


    public Session getSession() {
        return session;

    public void initIt() {
        cluster = Cluster.builder().addContactPoint(
        session = cluster.connect(keyspace);

    public void destroy() {
        if (session != null) {
        if (cluster != null) {



Automatically generated ...

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

public class BootCassandraDataShowApplication {

    public static void main(String[] args) {
        BootCassandraDataShowApplication.class, args);


Bean for holding configured objects.

package com.example;

import com.datastax.driver.core.Session;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

public class AppBeans {

    public Session session() {
        return sessionManager().getSession();

    public CassandraSessionManager sessionManager() {
        return new CassandraSessionManager();


Web Controller

package com.example;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Map;

public class WelcomeController {

    Session session;

    public String welcome(Map<String, Object> model) {

        final ResultSet rows = session.execute("SELECT * FROM votes");

        ArrayList results = new ArrayList<>();

        for (Row row : rows.all()) {
            results.add(new Vote(

        Collections.sort(results, (a, b) ->

        model.put("results", results);

        return "welcome";

Template to show the results

<!DOCTYPE html>
<html lang="en">

<h1>Voting results:</h1>
    <strong>{{this.name}}</strong> {{this.votes}} <br/>


That's all folks

Now this app might not seem as a lot, but there's a kafka cluster that receives messages comming in from a spring boot app that exposes REST interface. Messages that come in from kafka are then processed with Spark Streaming and then sent to Cassandra. There is another Spring Boot app that sorts and displays results to the users. This small tutorial covers most of the cool java/big data technologies now-days. Special thanks to the readers that went through all five parts of this tutorial ;)


Unknown said...

Hi Mark - This tutorial is really awesome, really appreciate your time and effort. It can really powerful. thanks

Marko Švaljek said...

Thank you for reading!

Akhil said...

Hi Mark ,

This is really awesome and I have one suggestion . If we can make this app as some analytics like streaming or feeding the data from twitter or so and streaming through web and pushing on to react front end.

But this is really awesome tutorial to get to start with the all the stack.


Marko Švaljek said...

The idea is great, I was also thinking about it for a while ... in the end decided to go for the simplest approach.

mrduguo said...

Interesting prototype. Do you mind share the sample code?

Thanks Marko for the brilliant work!

Marko Švaljek said...

password: hello

mrduguo said...

Excellent & thanks Marko for the prompt reply!

Best wishes!