Skip to content

Sinks

Jump to bottom Edit New page
mpollack edited this page Jun 11, 2013 · 28 revisions

Introduction

In this section we will show some variations on output sinks. As a prerequisite start the XD Container as instructed in the Getting Started page.

The Sinks covered are

See the section Creating a Sink Module for information on how to create sink modules using other Spring Integration Adapters.

Probably the simplest option for a sink is just to log the data. The log sink uses the application logger to output the data for inspection. The log level is set to WARN and the logger name is created from the stream name. To create a stream using a log sink you would use a command like

$ curl -d "http --port=8000 | log" http://localhost:8080/streams/mylogstream

You can then try adding some data. We’ve used the http source on port 8000 here, so run the following command to send a message

$ curl -d "hello" http://localhost:8000

and you should see the following output in the XD container console.

13/06/07 16:12:18 WARN logger.mylogstream: hello

The logger name is the sink name prefixed with the string "logger.". The sink name is the same as the stream name by default, but you can set it by passing the --name parameter

$ curl -d "http --port=8000 | log --name=mylogger" http://localhost:8080/streams/myotherlogstream

File Sink

Another simple option is to stream data to a file on the host OS. This can be done using the file sink module to create a stream.

$ curl -d "http --port=8000 | file" http://localhost:8080/streams/myfilestream

We’ve used the http source again, so run the following command to send a message

$ curl -d "hello" http://localhost:8000

The file sink uses the stream name as the default name for the file it creates, and places the file in the /tmp/xd/output/ directory.

$ less /tmp/xd/output/myfilestream
hello

You can cutomize the behavior and specify the name and dir properties of the output file. For example

$ curl -d "http --port=8000 | file --name=myfile --dir=/some/custom/directory" http://localhost:8080/streams/otherfilestream

Hadoop (HDFS)

First install and start Hadoop as described in our separate guide. It’s assumed HDFS is running on port 9000 (the default).

You should then be able to use the hdfs sink when creating a stream

$ curl -d "http --port=8000 | hdfs --rollover=10" http://localhost:8080/streams/myhdfsstream

Note that we’ve set the rollover parameter to a small value for this exercise. This is just to avoid buffering, so that we can actually see the data has made it into HDFS.

As in the above examples, we’ve used the http source on port 8000, so we can post some data again using

$ curl -d "hello" http://localhost:8000

Repeat the command a few times.

If you list the hadoop filesystem contents using hadoop fs -ls /, you should see that an xd directory has appeared in the root with a sub-directory named after our stream

$ hadoop dfs -ls /xd
Found 1 items
drwxr-xr-x   - luke supergroup          0 2013-05-28 14:53 /xd/myhdfsstream

And there will be one or more log files in there depending how many times you ran the command to post the data

$ hadoop dfs -ls /xd/myhdfsstream
Found 1 items
-rw-r--r--   3 luke supergroup          0 2013-05-28 14:53 /xd/myhdfsstream/myhdfsstream-0.log

You can examine the file contents using hadoop fs -cat

$ hadoop dfs -cat /xd/myhdfsstream/myhdfsstream-0.log
hello
hello
HDFS with Options

The HDFS Sink has the following options:

newline

whether to append a newline to the message payload (default: true)

directory

where to output the files in the Hadoop FileSystem (default: /xd/<streamname>)

filename

the base filename to use for the created files (a counter will be appended before the file extension). (default: <streamname>)

suffix

the file extension to use (default: log)

rollover

when to roll files over, expressed in bytes (default: 1000000, roughly 1MB)

TCP

The TCP Sink provides for outbound messaging over TCP.

The following examples use netcat (linux) to receive the data; the equivalent on Mac OSX is nc.

First, start a netcat to receive the data, and background it

$ netcat -l 1234 &

Now, configure a stream

$ curl -d "time --interval=3 | tcp" http://localhost:8080/streams/tcptest

This sends the time, every 3 seconds to the default tcp Sink, which connects to port 1234 on localhost.

$ Thu May 30 10:28:21 EDT 2013
Thu May 30 10:28:24 EDT 2013
Thu May 30 10:28:27 EDT 2013
Thu May 30 10:28:30 EDT 2013
Thu May 30 10:28:33 EDT 2013

TCP is a streaming protocol and some mechanism is needed to frame messages on the wire. A number of encoders are available, the default being 'CRLF'.

Undeploy the stream; netcat will terminate when the TCP Sink disconnects.

$ curl -X DELETE http://localhost:8080/streams/tcptest
TCP with Options

The TCP Sink has the following options

host

the host (or IP Address) to connect to (default: localhost)

port

the port on the host (default 1234)

reverse-lookup

perform a reverse DNS lookup on IP Addresses (default: false)

nio

whether or not to use NIO (default: false)

encoder

how to encode the stream - see below (default: CRLF)

close

whether to close the socket after each message (default: false)

charset

the charset used when converting text from String to bytes (default: UTF-8)

Retry Options

retry-max-attempts

the maximum number of attempts to send the data (default: 5 - original request and 4 retries)

retry-initial-interval

the time (ms) to wait for the first retry (default: 2000)

retry-multiplier

the multiplier for exponential back off of retries (default: 2)

With the default retry configuration, the attempts will be made after 0, 2, 4, 8, and 16 seconds.

Available Encoders
Text Data
CRLF (default)

text terminated by carriage return (0x0d) followed by line feed (0x0a)

LF

text terminated by line feed (0x0a)

NULL

text terminated by a null byte (0x00)

STXETX

text preceded by an STX (0x02) and terminated by an ETX (0x03)

Text and Binary Data
RAW

no structure - the client indicates a complete message by closing the socket

L1

data preceded by a one byte (unsigned) length field (supports up to 255 bytes)

L2

data preceded by a two byte (unsigned) length field (up to 216-1 bytes)

L4

data preceded by a four byte (signed) length field (up to 231-1 bytes)

An Additional Example

Start netcat in the background and redirect the output to a file foo

$ netcat -l 1235 > foo &

Create the stream, using the L4 encoder

$ curl -d "time --interval=3 | tcp --encoder=L4 --port=1235" http://localhost:8080/streams/tcptest

Undeploy the stream

$ curl -X DELETE http://localhost:8080/streams/tcptest

Check the output

$ hexdump -C foo
00000000  00 00 00 1c 54 68 75 20  4d 61 79 20 33 30 20 31  |....Thu May 30 1|
00000010  30 3a 34 37 3a 30 33 20  45 44 54 20 32 30 31 33  |0:47:03 EDT 2013|
00000020  00 00 00 1c 54 68 75 20  4d 61 79 20 33 30 20 31  |....Thu May 30 1|
00000030  30 3a 34 37 3a 30 36 20  45 44 54 20 32 30 31 33  |0:47:06 EDT 2013|
00000040  00 00 00 1c 54 68 75 20  4d 61 79 20 33 30 20 31  |....Thu May 30 1|
00000050  30 3a 34 37 3a 30 39 20  45 44 54 20 32 30 31 33  |0:47:09 EDT 2013|

Note the 4 byte length field preceding the data generated by the L4 encoder.

GemFire Server

Currently XD supports GemFire’s client-server topology. A sink that writes data to a GemFire cache requires a cache server to be running in a separate process and its host and port must be known (NOTE: GemFire locators are not supported yet). The XD distribution includes a GemFire server executable suitable for development and test purposes. It is made available under GemFire’s development license and is limited to 3 nodes. Modules that write to GemFire create a client cache and client region. No data is cached on the client.

Launching the XD GemFire Server

A GemFire Server is included in the Spring XD distribution. To start the server. Go to the XD install directory:

$cd gemfire/bin
$./gemfire-server cqdemo.xml

The command line argument is the location of a Spring file with a configured cache server. A sample cache configuration is provided cq-demo.xml. This starts a server on port 40404 and creates a region named Stocks. A Logging cache listener is configured for the region to log region events.

Gemfire sinks

There are 2 implementation of the gemfire sink: gemfire-server and gemfire-json-server. They are identical except the latter converts JSON string payloads to a JSON document format proprietary to GemFire and provides JSON field access and query capabilities. If you are not using JSON, the gemfire-server module will write the payload using java serialization to the configured region. Either of these modules accepts the following attributes:

regionName

the name of the GemFire region. This must be the name of a region configured for the cache server. This module creates the corresponding client region. (default: <streamname>)

keyExpression

A SpEL expression which is evaluated to create a cache key. Typically, the key value is derived from the payload. (default: <streamname>, which will overwrite the same entry for every message received on the stream)

gemfireHost

The host name or IP address of the cache server (default: localhost)

gemfirePort

The TCP port number of the cache server (default: 40404)

Example

Suppose we have a JSON document containing a stock price:

{"symbol":"VMW", "price":73}

We want this to be cached using the stock symbol as the key. The stream definition is:

http | gemfire-json-server --regionName=Stocks --keyExpression=payload.getField('symbol')

The keyExpression is a SpEL expression that depends on the payload type. In this case, com.gemstone.org.json.JSONObject. JSONObject which provides the getField method. To run this example:

$ curl -d "http --port=9090 | gemfire-json-server --regionName=Stocks --keyExpression=payload.getField('symbol')" http://localhost:8080/streams/stocks
$ curl -d "{\"symbol\":\"VMW\", \"price\":73}" http://localhost:9090

This will write an entry to the GemFire Stocks region with the key VMW. You should see a message on STDOUT for the process running the GemFire server like:

INFO [LoggingCacheListener] - updated entry VMW

Clone this wiki locally