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

Series

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

compile('com.datastax.cassandra:cassandra-driver-core:2.1.9')
    

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;
    }
}
    

application.properties

server.port = 8090
contactPoint = 127.0.0.1
keyspace = voting
    

CassandraSessionManager

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;

@Configuration
public class CassandraSessionManager {

    private Session session;
    private Cluster cluster;

    @Value("${contactPoint}")
    private String contactPoint;

    @Value("${keyspace}")
    private String keyspace;

    public CassandraSessionManager() {

    }

    public Session getSession() {
        return session;
    }

    @PostConstruct
    public void initIt() {
        cluster = Cluster.builder().addContactPoint(
            contactPoint).build();
        session = cluster.connect(keyspace);
    }

    @PreDestroy
    public void destroy() {
        if (session != null) {
            session.close();
        }
        if (cluster != null) {
            cluster.close();
        }
    }
}

    

BootCassandraDataShowApplication

Automatically generated ...

package com.example;

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

@SpringBootApplication
public class BootCassandraDataShowApplication {

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

AppBeans

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;

@Configuration
public class AppBeans {

    @Bean
    public Session session() {
        return sessionManager().getSession();
    }

    @Bean
    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;

@Configuration
@Controller
public class WelcomeController {

    @Autowired
    Session session;

    @RequestMapping("/")
    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(
                row.getString("name"),
                row.getInt("votes")
            ));
        }

        Collections.sort(results, (a, b) ->
        b.getVotes().compareTo(a.getVotes()));

        model.put("results", results);

        return "welcome";
    }
}
    

Template to show the results

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

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

</body>
</html>
    

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 ;)