#You can use Kafka’s Streams API or KSQL to easily process 
#our stream without any additional clusters to do some powerful 
#things - like get a top 5 chart of songs played. We’ll start by 
#joining a stream of songs that are played with an ID(playEvents) 
#and enriching that with songTable to get their name

final KStream songPlays =
  playEvents.leftJoin(songTable,(value1, song) -> song, 
  Serdes.Long(), playEventSerde);

#Next, we create a state store, ‘songPlayCounts’ to track song play counts

final KTable songPlayCounts =
  songPlays.groupBy((songId, song) -> song, keySongSerde, 
  valueSongSerde) .count(SONG_PLAY_COUNT_STORE);

final TopFiveSerde topFiveSerde = new TopFiveSerde();

#Lastly, we’ll compute the top five charts for each genre. 

#The results of this computation will continuously update

# the state which we can then query

songPlayCounts.groupBy((song, plays) ->
   KeyValue.pair(song.getGenre().toLowerCase(),
   new SongPlayCount(song.getId(), plays)),Serdes.String(),
   songPlayCountSerde)

#Kafka’s Connect API makes it simple to manage sources and sinks -

#for example this elasticsearch sink is an easy to prep

#friendly config file 

#User defined connector instance name.

name = elasticsearch-sink

#The class implementing the connector

connector.class = 
io.confluent.connect.elasticsearch.ElasticsearchSinkConnector

#Maximum number of tasks to run for this connector instance
tasks.max = 1
topics = test-elasticsearch-sink
key.ignore = true
connection.url = http://localhost:9200
type.name = kafka-connect

#Then once we’re set, we can run the elasticsearch connector

#in one line

$confluent load elasticsearch