6. Analytics

Create and deploy the following streams

The tweets stream subscribes to the provided twitter account, reads the incoming JSON tweets and logs their content to the log.

dataflow:>stream create tweets --definition "twitterstream --consumerKey=<CONSUMER_KEY> --consumerSecret=<CONSUMER_SECRET> --accessToken=<ACCESS_TOKEN> --accessTokenSecret=<ACCESS_TOKEN_SECRET> | log"

	Note
	To get a consumerKey and consumerSecret you need to register a twitter application. If you don’t already have one set up, you can create an app at the Twitter Developers site to get these credentials. The tokens <CONSUMER_KEY>, <CONSUMER_SECRET>, <ACCESS_TOKEN>, and <ACCESS_TOKEN_SECRET> are required to be replaced with your account credentials.

The received tweet messages would have a JSON format similar to this:

{
  "created_at": "Thu Apr 06 15:24:15 +0000 2017",
  "id_str": "850006245121695744",
  "text": "Today we are sharing our vision for the future of the Twitter API platform!",
  "user": {
    "id": 2244994945,
    "name": "Twitter Dev",
    "screen_name": "TwitterDev",
	"lang": "en"
  },
  "place": {},
  "entities": {
    "hashtags": [
		{
		  "text": "documentation",
		  "indices": [211, 225]
		},
		{
		  "text": "GeoTagged",
		  "indices": [239, 249]
		}
    ],
    ....
  }
}

The JsonPath SpEL expressions can help to extract the attributes to be analysed. For example the #jsonPath(payload,'$..lang') expression extracts all values of the lang attributes in the tweet. The Analytics Counter Sink maps the extracted values to custom Micrometer tags/dimensions attached to every measurement send. The tweetlang stream created below, extracts and counts the languages found in the tweets. The counter, named language, applies the --counter.tag.expression.lang=#jsonPath(payload,'$..lang') to extract the language values and map them to a Micrometer tag named: lang. This counter generates the language_total time-series send to Prometheus.

dataflow:>stream create tweetlang  --definition ":tweets.twitterstream > counter --counter.name=language --counter.tag.expression.lang=#jsonPath(payload,'$..lang')" --deploy

Similarly, we can use the #jsonPath(payload,'$.entities.hashtags[*].text') expression to extract and count the hastags in the incoming tweets. The following stream uses the counter-sink to compute real-time counts (named as hashtags) and the htag attribute in counter.tag.expression.htag indicate to Micrometer in what tag to hold the extracted hashtag values from the incoming tweets.

dataflow:>stream create tagcount  --definition ":tweets.twitterstream > counter --counter.name=hashtags --counter.tag.expression.htag=#jsonPath(payload,'$.entities.hashtags[*].text')" --deploy

Now we can deploy the tweets stream to start tweet analysis.

dataflow:>stream deploy tweets

Verify the streams are successfully deployed. Where: (1) is the primary pipeline; (2) and (3) are tapping the primary pipeline with the DSL syntax <stream-name>.<label/app name> [e.x. :tweets.twitterstream]; and (4) is the final deployment of primary pipeline

dataflow:>stream list