Introduction to Spring Cloud Dataflow (I)

Introduction

This is the first of a series of posts on how to develop data-driven micro-services with Spring Cloud Dataflow (SCDF from now on).  For now, we will see what is the approach proposed by this framework and how to build locally the basic components: Source, Sink and Processor.

Also, if you are familiar with Spring, we will take a look to the already-made components that are available for you to use so you don´t have to reinvent the wheel.

Contents

• What is Spring Cloud DataFlow?
• Introduction to the API: A simple Producer (Source), Consumer (sink) setup.
• Prerequisites: Kafka and Zookeeper.
• Coding a simple Source: HTTP Poller that retrieves live stock prices.
• Coding a simple Sink that store the results in a file.
• Writing an additional Batch Source.
• Summary & Resources.

Docker images creation in a development environment

Summary

Sorry, It´s been a while since my last post, a mixture of not having anything worth writing, holidays and some online training that drains most of my time.
Here is something I would like to share that might reduce the pain for those who want to dockerize their Spring-Boot applications (or any Java application) and then test and tune them while running inside a Docker container.

Former setup

• A commit was pushed to Jenkins, who compiled and unit-tested it.
• After some code analysis, a Docker image was created and pushed to a repository

With set-up, testing was highly inefficient.

• To test it in a server, I had to pull it and execute it and, while I was focused in the performance and memory consumption of my application, tuning the JVM parameters was essential... Unfortunately, they were given at compilation time (a Maven plugin) and could not be changed (or I did not know how to).

The drawbacks I see of this approach:

• It was slow to test new changes, as the whole CI flow needed to be executed.
• Unnecessary docker-related commits to the branch generating the Docker images.

The solution I found? Building the image myself before testing it.

Testing & error handling with Spring + Reactor

Introduction

As we saw in the previous post, the whole idea of reactive programming floats around two concepts: Data comes in asynchronous streams and those streams are immutable, so any modification results in new streams being created.
These facts makes the code not very intuitive and it is even worse when it comes to testing (even more if we are interacting with external resources, like a Web Service).
To help in the task, Spring and Reactor come with some handy tools that can be applied to simplify the development and testing.

First, let´s take a look at the scenario being tested:

Sequence diagram of the "production" code

Intro to Reactive Programming with Reactor and Spring

Introduction

What did I know last week about reactive programming? Nothing. That´s a fact. However, I started reading about and I find the topic really really interesting.

Unfortunately, there is not much documentation around (or at least, I had a hard time finding it) so I would like to share some resources with you along with an example that I have adapted to perform like our module Stokker (the module that used to fetch stock quotations from Yahoo Finance for us using Spring Integration).

What is Reactive Programming?

The best definition I could find was this amazing GitHub gist written by @andrestaltz that summarizes all the theoretical definitions to this line:

Reactive programming is programming with asynchronous data streams.

I really recommend to you its reading, but we can summarize it in:

• You will work with immutable event streams to which you will have to subscribe.
• Some operations will be provided to you, so you can transform these streams into new ones by filtering them, merging them o create new events from the previous ones.

The best way to understand this is, is to represent these operations graphically. For that purpose, we have this really cool web application implemented using JavaScript and RxObservables: RxMarbles (a must see!). For instance:

Two event streams: One with numbers, the other with each element multiplied by 10

The first line, represents an initial event stream. In the middle you can see the mapping function that is applied to every event in the stream and finally, the second stream with the results:

Note: Pay attention to the immutability here: Any transformation you apply to a stream creates a new one!

Let´s check out another transformation:

Two event streams: One with numbers, other with the average of the elements received

Look who is here! Our old friend, the moving average. Whenever a new event arrives in the original stream, a new value for the average will be calculated in the resulting stream.

Historical Stock Quotations as input for your Spring Application

Update on July 2nd 2017

Unfortunately, the API described in this article has been changed and now it requires a session cookie to access the data in CSV format. I guess the guys on Yahoo do not like to display their data without having the chance of displaying their ads...
Still, you might download the data manually (example URL here).
In the meantime, I´m looking for an alternate source of prices, I´ll keep you posted.

Summary

In a previous article, we defined how to get new stock quotations as soon as they were published in Google Finance (of course, after a 20-minute delay, this is a free service after all).
However, it was hard to study this data without having any historical data to compare with, so we defined a manual process to load some historical prices.

This manual process is very inefficient, though. You need to do it manually and there is a data gap between the last file update and the current quotations.

Solution: Yahoo finance historical quotations

We know that this service is not suitable for "real-time" quotations in case of the Spanish stock exchange, however, it is perfectly good for getting historical quotations during a period of time and offering more information such as:
- Volume
- Open, close, high and low prices
- Statistical data, such as mobile means, etc.

API Description

There seems to be no documentation for this interface at all (or at least, I have not been able to find it), however, I found some information about it buried in this 2009 blog article.

Basically, you need to compose a URL with the stock ticker and some time delimiters, for instance:

New Spring integration flow

As I had to replace the logic for reading the old static files, I created a new flow defined entirely in a separated XML. Basically what it does is:
- Reads from the application settings which stocks are we going to retrieve.
- Reads the time range to recover.
- Performs one request per Stock to the specific URL.
- Parses and converts the received CSV to our internal format.
- Publishes the new POJOs in to the topic so the  consumer can use this information seamlessly (it does not really care how the POJOs were built as long as the contract is respected).

Application.properties


HistoricalCSVRequestCreator

This class will receive a signal from Spring ApplicationContext as soon it is initialised (as it implements the interface ApplicationListener), it will parse the property containing the stocks to be retrieved and finally it will inject one message in the downstream channel per stock (Note that the ticker code is set as message header).


HTTP Outbound Gateway

This is the XML configuration for the component that will perform the request. Note how the target ticket is retrieved from the message header ticker and how the range of time is taken from the properties file.
And that´s it! As soon as you see the application running, a stream of data will be loaded into the ElasticSearch node so you can perform studies with wider statistical information.

As usual you can find the code in this GitHub repository. Feedback is always welcome!

Related articles

1. Live Stock data as input for your Spring application
2. Implement a Spring Integration flow with Spring Boot
3. Integrate the application with ElasticSearch
4. Represent and search your data with Kibana
5. Deploy your Spring Boot microservice in a Docker container

Integrate ElasticSearch with Spring-Boot

Summary

As commented in the previous post, the destinations for the recovered information will be:

- A simple log

- An Elasticsearch node, fur further data indexing and analysis.

How do we accomplish this? Very easy.

As we saw, during the project creation, we included the dependency spring-boot-starter-data-elasticsearch. This dependency will provide us with:

- A "template" API to perform  REST operations in towards an ES node, but without all the REST boilerplate.

- An embedded ES node, that works great in order to perform tests and help you understand ES.

Launching an ES node along with your Spring-Boot application:

If you want to launch an embedded ES node, you just need to:

• Enable it in your application .properties file by setting spring.data.elasticsearch.properties.http.enabled=true.
• Check that is properly running this curl command:
• $ curl -XGET 'http://localhost:9200/_cluster/health?pretty=true'
• Some cluster health statistics will be displayed (name, shards available, etc.)

Sending the information to ES

If you recall, our Spring Integration flow ended in a Service Activator. This component specified that @ELKClient.pushToELK()  would be invoked along with the payload of the message received (that is the CSV quotation). Well, let´s see the code used to perform the sending (Which I think is pretty much self-explanatory)

Building the document to be indexed

Finally we just need to create a simple bean, with some annotations specifying which fields are to be sent, analyzed and stored and in which format:

Next Steps

Once we run the application, it will load some files containing historical data and it will query the spreadsheets stored in Google Docs. Each entry will be converted to a Java Object and sent to Elastic Search. In the next post we will see how can we build some nice graphs using Kibana (part of ELK stack) and the data we have indexed. One example (showing the historical data):

Moreover, we will see how can we deploy the micro-service, the elastic search node and Kibana to the cloud, so we can continuously gather stock quotations and perform better analysis.

Source code

You can find the code in this GitHub repository.

Spring Boot/Integration gateway microservice

Summary

As we saw in the previous article, we created a stream of information coming from stock quotations that we will process in our application. What are the objectives? For the moment two:

• De-serialize the information and split it in tuples of (Stock,Price,Date)
• Publish these items to two streams
• A simple Log
• To the ELK stack (Elastic Search + Logstash + Kibana). We won´t use Logstash, though.

As usual with Spring Boot + Integration, the implementation of this kind of services is extremely easy.

Creation of the application  with Spring STS

Of course you can do this manually or by using a Maven archetype. However, Spring STS provides a handy wizard for creating applications with Spring-Boot that apart from creating the basic project framework, it allows you to add whatever Spring modules you may consider needed:

1. Modules chosen:
• Spring-Integration: Support for integration flows and usage of reusable components (file reading, messaging and tons more...)
•  Spring-Elastic Search (Cloud): This includes, among other features, a ES client that facilitates the sending of information towards ES. Moreover, using the default configuration, it launches a local ES node to play with.
2. I manually added two dependencies:
• Spring-Integration-File: This will allow us to load files with historical data
• Elastic Search: This is needed to work with the latest Kibana version (This will be explained more in depth in the next article)

For the sake of clarity, you can see how the pom.xml looks like here.

Implementation of the Integration business logic

All is accomplished by means of an XML defining the flow, a couple of Java classes and an import in the main class. This can be reduced even more by using more auto-configuration features.

• Enable the Spring Integration support for you XML file containing the flow

• Write the flow, in a file called /META-INF/feedFlow.xml, which will contain:

1. A http-outbound-gateway, that polls regularly the URL where the stock quotations are published
2. A file reader that reads the local files containing historical data.
3. A splitter, that creates one "item" per stock quotation.
4. A logger that logs each item
5. A service activator that sends the information to Elastic Search through REST

Let´s take a look to the most relevant elements with some more detail:

HTTP part of the flow

1. A int-http:outbound-gateway performs a HTTP request to the specified URL. The result of the request (the whole CSV file) is sent to the quotations channel
2. The request is triggered after an int:inbound-channel-adapter, which is driven by a poller, injects a message in the trigger channel.

File part of the flow

It might be useful, for further studies to have historical quotations. In order to process them a int-file:inbound-channel-adapter is polling a local folder where the user can place the files that he/she would like to load. These files are read and send to the quotations channel.

In this really simple example, we can see the powerful abstraction that Spring Integration gives, as we can inject quotations independently from where they come, as long as they respect the appropriate contract (format).

Final part of the flow

Finally, the CSV is split  in lines by a custom implementation of a Splitter and the result is made public in a publisher-subscriber channel, which is the Spring equivalent to a JMS topic.

Two consumers feed on these messages:

- A really simple logger, that logs the whole message (headers included)

- A service activator that send the quotations to the Elastic Search node.

We will see, how do we publish the data to ES and in which format in the next article. Moreover, we will create a couple of nice graphs using the data indexed in ES.

Source code

You can find the code in this GitHub repository.

I´m still improving things, like the error handling, configuration and proper conversion from CSV to the data expected by ES. In any case, feel free to share your critics/feedback.

Stock quotations as input for your Spring Integration App

Summary

As you might know Google deprecated its Google Finance API some time ago and other options such as Yahoo Finance API do not seem to work when dealing with Stocks outside the US (i.e. Spanish stocks are not supported).

So here I found an option of getting Stock prices by means of using a Google Spreadsheet as a source of information. Later on, we will consume this source of information with Spring Integration.

Creating and publishing your Google Spreadsheet containing stock quotations

If you are familiar with Google Docs and Spreadsheets, you will know for sure how simple is to create such documents, so I will omit the creation of the spreadsheet for the sake of clarity.

We will use the function GOOGLEFINANCE to retrieve the prices and timestamp of a number of stocks (there are many more features, you can see full function documentation is Google Spreadsheets) using three columns:

• Column A: Stock ticker
• Column B: Price
• The ticket will be taken from column A
• =GOOGLEFINANCE($A1)
• Column C:
• Last trade dateThe ticket will be taken from column A
• =GOOGLEFINANCE($A1;"tradetime")

With this setup, you will be able to write in column A as much stocks as you want and extend the formula to create a table. See this example (all but one are spanish stocks, OHI trades in the NYSE, note that for today we still don´t have a quotation).

Publish your sheet

Just go to File... Publish in the Web... and choose which sheets you want to publish and CSV format as output (of the formats available is the easiest by far to read later on by our application).

Check out this example here.

Next steps

After this simple set up, you will have a simple source of information with the stock parameters we need. In the next article, we will start building our Spring application (using Boot and Integration) to consume this source of data, store and analyze it.

Bear in mind that this free stock data is not suitable for automatic trading as it is not 100% reliable and usually come with a delay of around 20 minutes.

Update

This article refers to getting "live" stock data (i.e. delayed by 20 minutes). If you are also interested in getting historical stock data, Yahoo finance is still a good choice for you.  See this other article with more information on how to query the API provided by Yahoo Finance.