Dwell Dashboards On Streaming Information With Kinesis

We dwell in a world the place numerous techniques—social networks, monitoring, inventory exchanges, web sites, IoT gadgets—all constantly generate volumes of information within the type of occasions, captured in techniques like Apache Kafka and Amazon Kinesis. One can carry out all kinds of analyses, like aggregations, filtering, or sampling, on these occasion streams, both on the document degree or over sliding time home windows. On this weblog, I’ll present how Rockset can serve a dwell dashboard, which surfaces analytics on real-time Twitter information ingested into Rockset from a Kinesis stream.

Organising a Kinesis Stream

The Python code snippet beneath reveals tips on how to create a Kinesis stream programmatically. This can be achieved via the AWS Console or the AWS CLI.

import boto3
kinesis = boto3.consumer('kinesis') # requires AWS credentials to be current in env
kinesis.create_stream(StreamName="twitter-stream", ShardCount=5)

Writing Tweets to Kinesis

Right here, I will likely be utilizing the Tweepy module to fetch tweets via a streaming search API. This API permits me to specify an inventory of phrases that I wish to embody in my search (e.g. “music”, “fb”, “apple”). It’s good to have a Twitter developer account to be able to get entry to the Twitter Streaming API. Right here, I’ve a StreamListener, which is registered to be notified on a tweet arrival. Upon receiving a tweet, it writes the tweet information to one of many 5 random shards of the Kinesis stream.

# twitter api credentials
access_token=...
access_token_secret=...
consumer_key=...
consumer_secret=...

class TweetListener(StreamListener):
    def __init__(self, stream_name):
        self.kinesis = boto3.consumer('kinesis')
        self.stream_name = stream_name

    def on_data(self, information):
        document = {}
        document['Data'] = information
        document['PartitionKey'] = ''.be a part of(random.selection(chars) for _ in vary(dimension))
        self.kinesis.put_records(Data=[record], StreamName=self.stream_name)

auth=OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

stream=Stream(auth, TweetListener("twitter-stream"))
search_terms=["music", "facebook", "apple"]
stream.filter(monitor=search_terms)

Connecting Kinesis to Rockset

The next snippet reveals tips on how to create a group in Rockset, backed by a Kinesis stream. Be aware: It’s good to create an Integration (an object that represents your AWS credentials) and arrange related permissions on the Kinesis stream, which permits Rockset to carry out sure learn operations on that stream.

from rockset import Shopper, Q, F
rs=Shopper(api_key=...)

aws_integration=rs.Integration.retrieve(...)
sources=[
    rs.Source.kinesis(
        stream_name="twitter-stream",
        integration=aws_integration)]
twitter_kinesis_demo=rs.Assortment.create("twitter-kinesis-demo", sources=sources)

Alternatively, collections can be created from the Rockset console, as proven beneath.

Constructing the Dwell Dashboard

Now that I’ve a producer writing tweets to a Kinesis stream and a group to ingest them into Rockset, I can construct a dashboard on high of this assortment. My dashboard has two views.

Tweets View

The primary view shows analytics on all of the tweets coming into Rockset and has 3 panels, every of which makes its personal question to Rockset.

Dwell Tweets

The Dwell Tweets panel consistently refreshes to point out the most recent tweets showing within the assortment. A question is made at a set refresh interval to fetch tweets that had been tweeted within the final minute. Right here, I’m choosing required fields to point out on the feed and filtering out tweets older than a minute.

SELECT t.timestamp_ms,
   t.created_at AS created_at,
   t.textual content AS textual content,
   t.consumer.screen_name AS screen_name
FROM "twitter-kinesis-demo" t
WHERE CAST(timestamp_ms AS INT) > UNIX_MILLIS(current_timestamp() - minutes(1))
ORDER BY timestamp_ms DESC
LIMIT 100;

Prime Hashtags

The Prime Hashtags panel reveals trending hashtags, which had been present in most variety of tweets within the final hour, together with the related tweet depend. On this question, all hashtags showing within the final one hour are filtered into a short lived relation latest_hashtags. Utilizing a WITH clause, latest_hashtags is used it the primary question, the place we group by all of the hashtags and order by tweet_count to acquire the trending hashtags.

WITH lastest_hashtags AS
  (SELECT decrease(ht.textual content) AS hashtag
   FROM "twitter-kinesis-demo" t,
        unnest(t.extended_tweet.entities.hashtags) ht
   WHERE CAST(t.timestamp_ms AS INT) > UNIX_MILLIS(current_timestamp() - hours(1)))
SELECT depend(hashtag) AS tweet_count,
       hashtag
FROM latest_hashtags
GROUP BY hashtag
ORDER BY tweet_count DESC
LIMIT 10;

Incoming Tweets

The final panel is a chart which reveals the speed at which customers are tweeting. We get hold of information factors for the variety of incoming tweets each 2 seconds and plot them in a chart.

SELECT depend(*)
FROM "twitter-kinesis-demo"
WHERE forged(timestamp_ms AS INT) > unix_millis(current_timestamp() - seconds(2));

Hashtags View

The second view shows analytics on tweets with a particular hashtag and likewise has 3 panels: Dwell Tweets, Associated Hashtags, and Influencers. Every panel within the dashboard makes a question to Rockset. That is similar to the primary dashboard view however narrows the analytics to a specific hashtag of curiosity.

Influencers

As we’ve narrowed our evaluation to a single hashtag, it could be attention-grabbing to see who essentially the most influential customers are round this subject. For this, we outline influencers as customers with the best variety of followers who’re tweeting the hashtag of curiosity.

SELECT t.consumer.screen_name,
       t.consumer.followers_count AS fc
FROM "twitter-kinesis-demo" t
WHERE 'music' IN
    (SELECT hashtags.textual content
     FROM unnest(t.entities.hashtags) hashtags)
GROUP BY (t.consumer.screen_name,
          t.consumer.followers_count)
ORDER BY t.consumer.followers_count DESC
LIMIT 5;

Associated Hashtags

This part is considerably just like the Prime Hashtags panel we noticed within the Tweets view of the dashboard. It reveals the hashtags that co-occur most frequently together with our hashtag of curiosity.

SELECT hashtags.textual content as hashtag,
     depend(*) AS occurence_count
FROM "twitter-kinesis-demo" t,
    unnest(t.entites.hashtags) hashtags
WHERE 'music' IN
    (SELECT ht.textual content
     FROM unnest(t.entities.hashtags) ht)
  AND hashtags.textual content != 'music'
GROUP BY hashtags.textual content
ORDER BY occurence_count DESC
LIMIT 10;

Dwell Tweets

The Dwell Tweets panel is similar to one we noticed within the Tweets view of the dashboard. The one distinction is a brand new filter is utilized to be able to choose these tweets which comprise our hashtag of curiosity. I already used this filter for the opposite two panels within the Hashtags view.

The place to Go from Right here

Whereas I created this instance dwell dashboard for example how real-time analytics might be carried out on information from Kinesis streams, Rockset helps Kafka, as a streaming supply, and customary visualization instruments, like Tableau, Apache Superset, Redash, and Grafana, as properly.

You possibly can seek advice from the complete supply code for this instance right here, if you’re concerned with constructing on streaming information utilizing Rockset and Kinesis. Glad constructing!