Skip to content

Sinks

Jump to bottom Edit New page
David Turanski edited this page Jun 3, 2013 · 28 revisions

Sinks

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

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 -X POST -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.

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

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

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

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 -X POST -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'.

TCP with Options

The TCP Sink has the following options

  • host - the host (or IP Address) to connect to

  • 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 (default CRLF) - see below

  • 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

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

Available Encoders

Text Data

  • CRLF - 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 2**16-1 bytes)

  • L4 - data preceded by a four byte (signed) length field (up to 2**31-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 -X POST -d "time --interval=3 | tcp --encoder=L4 --port=1235" 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. The default value is the stream name.

  • keyExpression - A SpEL expression which is evaluated to create a cache key. Typically, the key value is derived from the payload. By default the cache key is the stream name, 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 -X POST -d "http --port=9090 | gemfire-json-server --regionName=Stocks --keyExpression=payload.getField('symbol')" http://localhost:8080/streams/stocks
$ curl -X POST -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