SpringCloud/spring-cloud-dataflow-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Twitter Analytics Demo with Micrometer Counters

Browse Source 
- Replace the old REDIS based example
 - add arch diagram

 Resolves #84
 Resolves https://github.com/spring-cloud/spring-cloud-dataflow/issues/2882
This commit is contained in:
Christian Tzolov
2019-02-20 13:38:44 +01:00
parent 19d9b41615
commit c101d448d9
6 changed files with 324 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
src/main/asciidoc/analytics/twitter-analytics/main.adoc
View File

@@ -3,7 +3,8 @@
:docs_dir: ../..
=== Twitter Analytics
In this demonstration, you will learn how to build a data pipeline using http://cloud.spring.io/spring-cloud-dataflow/[Spring Cloud Data Flow] to consume data from _TwitterStream_ and compute simple analytics over data-in-transit using _Field-Value-Counter_.
In this demonstration, you will learn how to build a data pipeline using http://cloud.spring.io/spring-cloud-dataflow/[Spring Cloud Data Flow] to consume data from _TwitterStream_, compute analytics over data-in-transit using https://github.com/spring-cloud-stream-app-starters/analytics[Analytics-Counter].
Use Prometheus for storing and data aggregation analysis and Grafana for visualizing the computed data.
We will take you through the steps to configure Spring Cloud Data Flow's `Local` server.
@@ -13,7 +14,15 @@ We will take you through the steps to configure Spring Cloud Data Flow's `Local`
include::{docs_dir}/shell.adoc[]
* A running local Data Flow Server
include::{docs_dir}/local-server.adoc[]
* Running instance of link:http://redis.io/[Redis]
Make sure to add the following properties when starting the Data Flow server:
```
--spring.cloud.dataflow.applicationProperties.stream.management.metrics.export.prometheus.enabled=true
--spring.cloud.dataflow.applicationProperties.stream.spring.cloud.streamapp.security.enabled=false
--spring.cloud.dataflow.applicationProperties.stream.management.endpoints.web.exposure.include=prometheus,info,health
--spring.cloud.dataflow.grafana-info.url=http://localhost:3000
```
* Running instance of link:http://docs.spring.io/spring-cloud-dataflow/docs/2.0.0.BUILD-SNAPSHOT/reference/htmlsingle/#streams-monitoring-local-prometheus[Prometheus, Service Discovery and Grafana].
Follow the http://docs.spring.io/spring-cloud-dataflow/docs/2.0.0.BUILD-SNAPSHOT/reference/htmlsingle/#streams-monitoring-local-prometheus[instructions] to start those services in Docker containers.
* Running instance of link:http://kafka.apache.org/downloads.html[Kafka]
* Twitter credentials from link:https://apps.twitter.com/[Twitter Developers] site
@@ -31,119 +40,102 @@ dataflow:>app import --uri {app-import-kafka-maven}
. Create and deploy the following streams
+
image::scdf-tweets-analysis-architecture.png[Twitter Analytics Visualization, scaledwidth="100%"]
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"
Created new stream 'tweets'
```
The received https://developer.twitter.com/en/docs/tweets/data-dictionary/overview/intro-to-tweet-json.html[tweet messages] have a format similar to this:
+
[source,json]
----
{
"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 https://github.com/json-path/JsonPath[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 https://github.com/spring-cloud-stream-app-starters/analytics/tree/master/spring-cloud-starter-stream-sink-counter[Analytics Counter Sink] maps the extracted values to custom https://micrometer.io/docs/concepts#_meters[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 > field-value-counter --fieldName=lang --name=language" --deploy
dataflow:>stream create tweetlang --definition ":tweets.twitterstream > counter --name=language --counter.tag.expression.lang=#jsonPath(payload,'$..lang')" --deploy
Created and deployed new stream 'tweetlang'
```
+
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 > field-value-counter --fieldName=entities.hashtags.text --name=hashtags" --deploy
dataflow:>stream create tagcount --definition ":tweets.twitterstream > counter --name=hashtags --counter.tag.expression.htag=#jsonPath(payload,'$.entities.hashtags[*].text')" --deploy
Created and deployed new stream 'tagcount'
```
+
Now we can deploy the `tweets` stream to start tweet analysis.
+
```
dataflow:>stream deploy tweets
Deployed stream 'tweets'
```
+
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 link:https://apps.twitter.com/[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.
+
. 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
```
+
. Notice that `tweetlang.field-value-counter`, `tagcount.field-value-counter`, `tweets.log` and `tweets.twitterstream` link:https://github.com/spring-cloud-stream-app-starters/[Spring Cloud Stream] applications are running as Spring Boot applications within the `local-server`.
. Notice that `tweetlang.counter`, `tagcount.counter`, `tweets.log` and `tweets.twitterstream` link:https://github.com/spring-cloud-stream-app-starters/[Spring Cloud Stream] applications are running as Spring Boot applications within the `local-server`.
+
[source,console,options=nowrap]
----
2016-02-16 11:43:26.174 INFO 10189 --- [nio-9393-exec-2] o.s.c.d.d.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:field-value-counter-sink:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-6990537012958280418/tweetlang-1455651806160/tweetlang.field-value-counter
2016-02-16 11:43:26.206 INFO 10189 --- [nio-9393-exec-3] o.s.c.d.d.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:field-value-counter-sink:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-6990537012958280418/tagcount-1455651806202/tagcount.field-value-counter
2016-02-16 11:43:26.806 INFO 10189 --- [nio-9393-exec-4] o.s.c.d.d.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:log-sink:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-6990537012958280418/tweets-1455651806800/tweets.log
2016-02-16 11:43:26.813 INFO 10189 --- [nio-9393-exec-4] o.s.c.d.d.l.OutOfProcessModuleDeployer : deploying module org.springframework.cloud.stream.module:twitterstream-source:jar:exec:1.0.0.BUILD-SNAPSHOT instance 0
Logs will be in /var/folders/c3/ctx7_rns6x30tq7rb76wzqwr0000gp/T/spring-cloud-data-flow-6990537012958280418/tweets-1455651806800/tweets.twitterstream
----
+
. Verify that two `field-value-counter` with the names `hashtags` and `language` is listing successfully
+
[source,console,options=nowrap]
----
dataflow:>field-value-counter list
╔════════════════════════╗
║Field Value Counter name║
╠════════════════════════╣
║hashtags ║
║language ║
╚════════════════════════╝
----
+
. Verify you can query individual `field-value-counter` results successfully
+
[source,console,options=nowrap]
----
dataflow:>field-value-counter display hashtags
Displaying values for field value counter 'hashtags'
╔══════════════════════════════════════╤═════╗
║ Value │Count║
╠══════════════════════════════════════╪═════╣
║KCA │ 40║
║PENNYSTOCKS │ 17║
║TEAMBILLIONAIRE │ 17║
║UCL │ 11║
║... │ ..║
║... │ ..║
║... │ ..║
╚══════════════════════════════════════╧═════╝
dataflow:>field-value-counter display language
Displaying values for field value counter 'language'
╔═════╤═════╗
║Value│Count║
╠═════╪═════╣
║en │1,171║
║es │ 337║
║ar │ 296║
║und │ 251║
║pt │ 175║
║ja │ 137║
║.. │ ...║
║.. │ ...║
║.. │ ...║
╚═════╧═════╝
----
+
. Go to `Dashboard` accessible at `http://localhost:9393/dashboard` and launch the `Analytics` tab. From the default `Dashboard` menu, select the following combinations to visualize real-time updates on `field-value-counter`.
- For real-time updates on `language` tags, select:
.. Metric Type as `Field-Value-Counters`
.. Stream as `language`
.. Visualization as `Bubble-Chart` or `Pie-Chart`
- For real-time updates on `hashtags` tags, select:
.. Metric Type as `Field-Value-Counters`
.. Stream as `hashtags`
.. Visualization as `Bubble-Chart` or `Pie-Chart`
. Go to `Grafana Dashboard` accessible at `http://localhost:3000`, login as admin:admin.
Import the link:micrometer/prometheus/grafana-twitter-scdf-analytics.json[grafana-twitter-scdf-analytics.json] dashboard.
You will see a dashboard similar to this:
image::twitter_analytics.png[Twitter Analytics Visualization, scaledwidth="50%"]
The following Prometheus queries have been used to aggregate the `lang` and `htag` data persisted in Prometheus, which can be visualized through Grafana dashboard:
[source,console,options=nowrap]
----
sort_desc(topk(10, sum(language_total) by (lang)))
sort_desc(topk(100, sum(hashtags_total) by (htag)))
----
==== Summary
In this sample, you have learned:
* How to use Spring Cloud Data Flow's `Local` server
* How to use Spring Cloud Data Flow's `shell` application
* How to create streaming data pipeline to compute simple analytics using `Twitter Stream` and `Field Value Counter` applications
* How to use Prometheus and Grafana with Spring Cloud Data Flow's `Local` server
* How to create streaming data pipeline to compute simple analytics using `Twitter Stream` and `Analytics Counter` applications

BIN
src/main/asciidoc/images/scdf-tweets-analysis-architecture.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 85 KiB

BIN
src/main/asciidoc/images/twitter_analytics.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 416 KiB

After

Width:  |  Height:  |  Size: 740 KiB

2
src/main/asciidoc/index.adoc
View File

@@ -11,7 +11,7 @@ Sabby Anandan; David Turanski; Glenn Renfro; Eric Bottard; Mark Pollack; Chris S
:spring-cloud-stream-docs: http://docs.spring.io/spring-cloud-stream/docs/{scst-core-version}/reference/htmlsingle/index.html
:github-code: https://github.com/spring-cloud/spring-cloud-dataflow-samples
:scdf-release-train-name: Celsius-SR1
:scdf-release-train-name: Einstein.RELEASE
:app-import-kafka-maven: http://bit.ly/{scdf-release-train-name}-stream-applications-kafka-10-maven
:app-import-rabbit-maven: http://bit.ly/{scdf-release-train-name}-stream-applications-rabbit-maven

4
src/main/asciidoc/maven-access.adoc
View File

@@ -2,6 +2,6 @@ NOTE: These samples assume that the Data Flow Server can access a remote Maven r
access to public repositories, you will need to install the sample apps in your internal Maven repository and https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#getting-started-maven-configuration[configure]
the server accordingly. The sample applications are typically registered using Data Flow's bulk import facility. For example, the Shell command `dataflow:>app import --uri {app-import-rabbit-maven}` _(The actual URI is release and binder specific so refer to the sample instructions for the actual URL)_.
The bulk import URI references a plain text file containing entries for all of the publicly available Spring Cloud Stream and Task applications published to `https://repo.spring.io`. For example,
`source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:1.3.1.RELEASE` registers the `http` source app at the corresponding Maven address, relative to the remote repository(ies) configured for the
Data Flow server. The format is `maven://<groupId>:<artifactId>:<version>` You will need to https://repo.spring.io/libs-release/org/springframework/cloud/stream/app/spring-cloud-stream-app-descriptor/Bacon.RELEASE/spring-cloud-stream-app-descriptor-Bacon.RELEASE.rabbit-apps-maven-repo-url.properties[download] the required apps or https://github.com/spring-cloud-stream-app-starters[build] them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do
`source.http=maven://org.springframework.cloud.stream.app:http-source-rabbit:2.1.0.RELEASE` registers the `http` source app at the corresponding Maven address, relative to the remote repository(ies) configured for the
Data Flow server. The format is `maven://<groupId>:<artifactId>:<version>` You will need to https://repo.spring.io/libs-release/org/springframework/cloud/stream/app/spring-cloud-stream-app-descriptor/Einstein.RELEASE/spring-cloud-stream-app-descriptor-Einstein.RELEASE.rabbit-apps-maven-repo-url.properties[download] the required apps or https://github.com/spring-cloud-stream-app-starters[build] them and then install them in your Maven repository, using whatever group, artifact, and version you choose. If you do
this, register individual apps using `dataflow:>app register...` using the `maven://` resource URI format corresponding to your installed app.

247
src/main/asciidoc/micrometer/prometheus/grafana-twitter-scdf-analytics.json Normal file
View File

@@ -0,0 +1,247 @@
{
"annotations": {
"list": [
{
"builtIn": 1,
"datasource": "-- Grafana --",
"enable": true,
"hide": true,
"iconColor": "rgba(0, 211, 255, 1)",
"name": "Annotations & Alerts",
"type": "dashboard"
}
]
},
"editable": true,
"gnetId": null,
"graphTooltip": 0,
"id": 3,
"links": [
{
"icon": "external link",
"tags": [],
"type": "dashboards"
}
],
"panels": [
{
"aliasColors": {},
"bars": true,
"dashLength": 10,
"dashes": false,
"fill": 1,
"gridPos": {
"h": 10,
"w": 15,
"x": 0,
"y": 0
},
"id": 6,
"legend": {
"alignAsTable": false,
"avg": false,
"current": false,
"max": false,
"min": false,
"show": false,
"total": false,
"values": false
},
"lines": false,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pointradius": 5,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "sort_desc(topk(10, sum(language_total) by (lang)))",
"format": "time_series",
"instant": true,
"intervalFactor": 1,
"legendFormat": "{{lang}}",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Top 10 Lang",
"tooltip": {
"shared": false,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "series",
"name": null,
"show": true,
"values": [
"current"
]
},
"yaxes": [
{
"format": "short",
"label": "",
"logBase": 1,
"max": null,
"min": "0",
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": false
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"aliasColors": {},
"breakPoint": "50%",
"cacheTimeout": null,
"combine": {
"label": "Others",
"threshold": 0
},
"fontSize": "80%",
"format": "short",
"gridPos": {
"h": 22,
"w": 9,
"x": 15,
"y": 0
},
"id": 4,
"interval": null,
"legend": {
"show": true,
"values": true
},
"legendType": "Under graph",
"links": [],
"maxDataPoints": 3,
"nullPointMode": "connected",
"pieType": "donut",
"strokeWidth": 1,
"targets": [
{
"expr": "sort_desc(topk(50, sum(hashtags_total) by (htag)))",
"format": "time_series",
"instant": true,
"intervalFactor": 1,
"legendFormat": "{{htag}}",
"refId": "A"
}
],
"title": "Hastags",
"type": "grafana-piechart-panel",
"valueName": "current"
},
{
"bgColor": null,
"colorScheme": "Unique",
"decimal": 2,
"displayLabel": true,
"format": "short",
"gradientColors": [
"red",
"green"
],
"gradientThresholds": "50,80",
"gridPos": {
"h": 12,
"w": 15,
"x": 0,
"y": 10
},
"groupDepthColors": [
"hsl(152,80%,80%)",
"hsl(228,30%,40%)"
],
"groupSeperator": ",",
"height": 400,
"id": 2,
"links": [],
"mode": "time",
"nullPointMode": "connected",
"svgBubbleId": "svg_2",
"svgContainer": {},
"targets": [
{
"expr": "sum(language_total) by (lang)",
"format": "time_series",
"instant": true,
"intervalFactor": 1,
"legendFormat": "{{lang}}",
"refId": "A"
}
],
"thresholdColors": [
"green",
"yellow",
"red"
],
"thresholds": "50,80",
"title": "Languages",
"type": "digrich-bubblechart-panel",
"valueName": "current"
}
],
"refresh": "5s",
"schemaVersion": 16,
"style": "dark",
"tags": [],
"templating": {
"list": []
},
"time": {
"from": "now-30m",
"to": "now"
},
"timepicker": {
"refresh_intervals": [
"5s",
"10s",
"30s",
"1m",
"5m",
"15m",
"30m",
"1h",
"2h",
"1d"
],
"time_options": [
"5m",
"15m",
"1h",
"6h",
"12h",
"24h",
"2d",
"7d",
"30d"
]
},
"timezone": "",
"title": "SCDF Analytics",
"uid": "vhHweSriz",
"version": 2
}