Skip to content

Sinks

Jump to bottom Edit New page
Gary Russell edited this page May 30, 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)

Install Hadoop and start it using the start-all.sh script. 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 bufffering, 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 fs -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 fs -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 fs -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.

Clone this wiki locally