Taps

Taps

A tap acts like a source in that it occurs as the first module within a stream and can pipe its output to a sink (and/or one or more processors added to a chain before the ultimate sink), but for a tap the messages are actually those being produced by some other source. Each tap consumes a copy of every incoming message on the target stream so it can process the same data at the same time without affecting the target stream. You may think of a tap as a "tee" stream.

A primary use case is to perform realtime analytics at the same time as data is being ingested via its primary stream. Typically a counter or gauge would follow the pipe after a tap. The syntax for creating a tap is:

tap @ <target stream>

In this section we will show some examples of creating taps on existing streams. As a prerequisite start the XD Container as instructed in the Getting Started page.

The Taps covered are

• Counter

• Field Value Counter

Counter

A counter is a Metric that associates a unique name with a long value. It is primarily used for counting events triggered by incoming messages on a target stream. You create a counter with a unique name and optionally an initial value then set its value in response to incoming messages. The most straightforward use for counter is simply to count messages coming into the target stream. That is, its value is incremented on every message. This is exactly what the counter module provided by Spring XD does.

Here’s an example:

Start by creating a main stream to perform data ingest. Something like:

$curl -X POST -d "twittersearch --query=spring | file --directory=/tweets/" http://localhost:8080/streams/springtweets

Next, create a tap on the springtweets stream that sets a message counter named tweetcount

$curl -X POST -d "tap @ springtweets | counter --name=tweetcount" http://localhost:8080/streams/tweettap

The results are written to redis under the key counter.${name}. To retrieve the count:

$redis-cli
$redis 127.0.0.1:6379> get counters.tweetcount

Field Value Counter

A field value counter is a Metric used for counting occurrences of unique values for a named field in a message payload. XD Supports the following payload types out of the box:

• POJO (Java bean)

• Tuple

• JSON String

For example suppose a message source produces a payload with a field named user :

class Foo {
   String user;
   public Foo(String user) {
       this.user = user;
   }
}

If the stream source produces messages with the following objects:

   new Foo("fred")
   new Foo("sue")
   new Foo("dave")
   new Foo("sue")

The field value counter on the field user will contain:

fred:1, sue:2, dave:1

Multi-value fields are also supported. For example, if a field contains a list, each value will be counted once:

users:["dave","fred","sue"]
users:["sue","jon"]

The field value counter on the field users will contain:

dave:1, fred:1, sue:2, jon:1

field_value_counter has the following options:

• fieldName: The name of the field for which values are counted (required)

• counterName: A key used to access the counter values. This defaults to ${fieldName}.

To try this out, create a stream to ingest twitter feeds containing the word spring and output to a file:

curl -X POST -d "twittersearch --query=spring | file" http://localhost:8080/streams/springtweets

Now create a tap for a field value counter:

curl -X POST -d "tap@springtweets | field-value-counter --fieldName=fromUser" http://localhost:8080/streams/tweettap

The twittersearch source produces JSON strings which contain the user id of the tweeter in the fromUser field. The field_value_counter sink parses the tweet and updates a field value counter named fromUser in Redis. To view the counts:

$redis-cli
redis 127.0.0.1:6379>zrange fieldvaluecounters.fromUser 0 -1 withscores