Skip to content

Streams

Jump to bottom Edit New page
Luke Taylor edited this page May 23, 2013 · 16 revisions

Introduction

In Spring XD, a basic stream defines the ingestion of data from a source to a sink. Streams are managed using the StreamServer from the spring-xd-dirt subproject and defined in terms of source and sink modules. Modules are predefined configurations for sources and sinks and examples can be found in the modules directory within the spring-xd project [1]. When you ask the StreamServer to set up a new stream, it looks in the modules directory for the source and sink definitions.

The StreamServer

The StreamServer is a prototype implementation for managing streams. TODO: XD-Server + other functionality. You can start it by running ./gradlew launch from the root directory. It exposes a REST API (at the /streams) endpoint to which you can post stream definition commands to define new streams. The stream definitions are built from a simple DSL. For example, let’s walk through what happens if we execute the following request

$ curl -X POST -d "testsource | testsink" http://localhost:8080/streams/myteststream

The DSL uses the "pipe" symbol |, to connect a source to a sink. The stream server finds the testsource and testsink definitions in the modules directory and uses them to setup the stream [2]. In this simple example, the testsource simply sends a single "hello" message, and the testsink outputs it using the logging framework.

processing module 'Module [name=testsink, type=sink]' from group 'example' with index: 1
processing module 'Module [name=testsource, type=source]' from group 'example' with index: 0
16:11:50,542  WARN ThreadPoolTaskScheduler-1 handler.LoggingHandler:141 - hello

Other Source and Sink Types

Let’s try something a bit more complicated and swap out the testsource for something else. Another supported source type is http, which accepts data for ingestion over HTTP POSTs. Note that this should not be confused with the REST API which the server uses to create streams. The https source accepts data on a different port (default 9000), from the API one.

To create a stream using an http source, but still using the same testsink, we would change the original command above to

$ curl -X POST -d "http | testsink" http://localhost:8080/streams/myhttpteststream

which will produce the following output from the server

processing module 'Module [name=testsink, type=sink]' from group 'myhttpteststream' with index: 1
processing module 'Module [name=http, type=source]' from group 'myhttpteststream' with index: 0

Note that we don’t see any other output this time until we actually post some data

$ curl -X POST -d "hello"
$ curl -X POST -d "goodbye"

and the stream will then funnel the data from the http source to the output log implemented by the testsink

15:08:01,676  WARN ThreadPoolTaskScheduler-1 handler.LoggingHandler:141 - hello
15:08:12,520  WARN ThreadPoolTaskScheduler-1 handler.LoggingHandler:141 - goodbye

Of course, we could also change the sink implementation. You could pipe the output to a file (file), to hadoop (hdfs) or to any of the other sink modules which are provided. You can also define your own modules TODO: Link.

DSL Syntax

In the examples above, we connected a source to a sink using the pipe symbol |. You can also pass parameters to the source and sink configurations. The parameter names will depend on the individual module implementations, but as an example, the http source module exposes a port setting which allows you to change the data ingestion port from the default value. To create the stream using port 8000, we would use

$ curl -X POST -d "http --port=8000 | testsink" http://localhost:8080/streams/myhttpteststream

If you know a bit about Spring configuration files, you can inspect the module definition to see which properties it exposes. Alternatively, you can read more in the module documentation TODO: Link to source and sink docs???.

Questions I Would Have as a Beginner

What does REDIS do in all this?

How do I control the lifecycle of streams - e.g. if I have several running and want to stop one. Or if I accidentally create one with the wrong configuration and want to replace it.

What does the group refer to? e.g. in processing module 'Module [name=testsink, type=sink]' from group 'myhttpteststream'

1. Using the filesystem is just one possible way of configuring a module registry. A Redis implementation would be another option.
2. The definitions are just standard Spring configuration files

Clone this wiki locally