Real-Time Data Analysis with Kubernetes, Redis, and BigQuery

Creating your BigQuery table

Create a BigQuery table to store your tweets. BigQuery groups tables into abstraction layers called datasets. Use the bq command-line tool, which is included with the Cloud SDK, to create a new BigQuery dataset named rtda.

1. In a terminal window, enter the following command:

   bq mk rtda
   

2. Now that you have a dataset in place, create a new table named tweets to contain the incoming tweets. Each BigQuery table must be defined by a schema. To save you time, this example provides a predefined schema, schema.json, that you can use to define your table. To create the table by using the predefined schema, enter the following command:

   bq mk -t rtda.tweets kube-redis-bq/bigquery-setup/schema.json
   

3. To verify that your new dataset and table have been created, use the BigQuery web UI. You should see the dataset name in the left sidebar. If it isn't there, make sure you're looking at the correct project. When you click the arrow next to the dataset name, you should see your new table. Alternatively, you can view all datasets in a project or all tables in a given dataset by running the following commands:

   bq ls [PROJECT_ID]:
   

   bq ls [DATASET_ID]
   

   In this case, the dataset ID is rtda.

4. Finally, edit kube-redis-bq/redis/bigquery-controller.yaml. Update the value for the following fields to reflect your BigQuery configuration:

   • PROJECT_ID: Your project ID.
   • BQ_DATASET: The BigQuery dataset containing your table (rtda).
   • BQ_TABLE: The BigQuery table you just created (tweets).

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: bigquery-controller
  labels:
    name: bigquery-controller
spec:
  replicas: 2
  template:
    metadata:
      labels:
        name: bigquery-controller
    spec:
      containers:
      - name: bigquery
        image: gcr.io/google-samples/redis-bq-pipe:v3
        env:
        - name: PROCESSINGSCRIPT
          value: redis-to-bigquery
        - name: REDISLIST
          value: twitter-stream
        # Change this to your project ID.
        - name: PROJECT_ID
          value: xxxx
        # Change the following two settings to your dataset and table.
        - name: BQ_DATASET
          value: xxxx
        - name: BQ_TABLE
          value: xxxx

Configuring and starting a Kubernetes cluster

Kubernetes is portable—it uses a set of shell scripts for starting up, shutting down, and managing configuration—so no installation is necessary.

1. Edit kubernetes/cluster/gce/config-common.sh.

2. Add the bigquery scope to the file's NODE_SCOPES definition.

This setting allows your nodes to write to your BigQuery table. Save the configuration and close your file.

(Optional) Update the startup scripts for long-term Redis reliability

You can improve the long-term reliability of Redis on Kubernetes by adding some suggested optimizations to the nodes' startup scripts. This step is helpful if you want your Redis master to run for a long period of time. If you don't intend to have your cluster up for very long, you can skip this step.

1. Edit kubernetes/cluster/gce/configure-vm.sh and scroll to the bottom of the file. Above the following if statement, after the row of # symbols:

   if [[ -z "${is_push}" ]]; then
   

2. Insert the following commands:

   /sbin/sysctl vm.overcommit_memory=1
   echo never > /sys/kernel/mm/transparent_hugepage/enabled
   

These commands will set your Linux kernel to always overcommit memory and will disable transparent huge pages in the kernel. Save and close the file.

Start up your cluster

Now you are ready to start up your cluster.

1. Enter the following command:

   kubernetes/cluster/kube-up.sh
   

   Starting the cluster can take some time. During the startup process, you might be prompted to create a new SSH key for Compute Engine or, if you've already done so previously, to enter your SSH key passphrase.

2. After your cluster has started, enter the following commands to see your running instances:

   kubectl get nodes

   kubectl cluster-info

You should see one Kubernetes master instance and four Kubernetes nodes.

Creating a Twitter application

To receive tweets from Twitter, you need to create a Twitter application and add its key/secret and token/secret values to the specification for your Twitter-to-Redis Kubernetes pod. In all, you will copy four values. Follow these steps:

1. Create a new Twitter application.
2. In the Twitter Application Management page, navigate to the Keys and Access Tokens tab.
3. Click the Create my access token button to create a new access token.
4. Edit kube-redis-bq/redis/twitter-stream.yaml.
5. Replace the following values with your consumer key and consumer secret:
   • CONSUMERKEY
   • CONSUMERSECRET
6. Replace the following values with your access token and access token secret:
   • ACCESSTOKEN
   • ACCESSTOKENSEC

The example application reads from a random sample of Twitter's public stream by default. To filter on a set of keywords instead:

1. Edit kube-redis-bq/twitter-stream.yaml and change the value of TWSTREAMMODE to filter.
2. Edit twitter-to-redis.py and replace the keywords defined in the track variable with the keywords you want to filter on.
3. Rebuild the container image. For instructions about how to build the container image, see the appendix.

Understanding the Docker image for your Kubernetes pods

This example uses three different Kubernetes pod templates to construct the analysis pipeline. One is a Redis pod, and the other two types of pods run the example application's scripts.

The specification files point to a pre-built Docker image; you don't need to do anything special to use it. If you'd like to build the image yourself, follow the steps in the appendix.

The Docker image contains two main scripts that perform the work for this solution.

The code in twitter-to-redis.py streams incoming tweets from Twitter to Redis.

#!/usr/bin/env python
# Copyright 2014 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""This script uses the Twitter Streaming API, via the tweepy library,
to pull in tweets and store them in a Redis server.
"""

import datetime
import os

import redis
from tweepy import OAuthHandler
from tweepy import Stream
from tweepy.streaming import StreamListener

# Get your twitter credentials from the environment variables.
# These are set in the 'twitter-stream.json' manifest file.
consumer_key = os.environ['CONSUMERKEY']
consumer_secret = os.environ['CONSUMERSECRET']
access_token = os.environ['ACCESSTOKEN']
access_token_secret = os.environ['ACCESSTOKENSEC']

# Get info on the Redis host and port from the environment variables.
# The name of this variable comes from the redis service id, 'redismaster'.
REDIS_HOST = os.environ['REDISMASTER_SERVICE_HOST']
REDIS_PORT = os.environ['REDISMASTER_SERVICE_PORT']
REDIS_LIST = os.environ['REDISLIST']


class StdOutListener(StreamListener):
    """A listener handles tweets that are received from the stream.
    This listener dumps the tweets into Redis.
    """

    count = 0
    redis_errors = 0
    allowed_redis_errors = 3
    twstring = ''
    tweets = []
    r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=0)
    total_tweets = 10000000

    def write_to_redis(self, tw):
        try:
            self.r.lpush(REDIS_LIST, tw)
        except:
            print 'Problem adding data to Redis.'
            self.redis_errors += 1

    def on_data(self, data):
        """What to do when tweet data is received."""
        self.write_to_redis(data)
        self.count += 1
        # if we've grabbed more than total_tweets tweets, exit the script.
        # If this script is being run in the context of a kubernetes
        # replicationController, the pod will be restarted fresh when
        # that happens.
        if self.count > self.total_tweets:
            return False
        if self.redis_errors > self.allowed_redis_errors:
            print 'too many redis errors.'
            return False
        if (self.count % 1000) == 0:
            print 'count is: %s at %s' % (self.count, datetime.datetime.now())
        return True

    def on_error(self, status):
        print status


if __name__ == '__main__':
    print '....'
    listener = StdOutListener()
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    print 'stream mode is: %s' % os.environ['TWSTREAMMODE']

    stream = Stream(auth, listener)
    # set up the streaming depending upon whether our mode is 'sample', which
    # will sample the twitter public stream. If not 'sample', instead track
    # the given set of keywords.
    # This environment var is set in the 'twitter-stream.yaml' file.
    if os.environ['TWSTREAMMODE'] == 'sample':
        stream.sample()
    else:
        stream.filter(
                track=['bigdata', 'kubernetes', 'bigquery', 'docker', 'google',
                       'googlecloud', 'golang', 'dataflow',
                       'containers', 'appengine', 'gcp', 'compute',
                       'scalability', 'gigaom', 'news', 'tech', 'apple',
                       'amazon', 'cluster', 'distributed', 'computing',
                       'cloud', 'android', 'mobile', 'ios', 'iphone',
                       'python', 'recode', 'techcrunch', 'timoreilly']
                )

The code in redis-to-bigquery.py streams cached data from Redis to BigQuery.

#!/usr/bin/env python
# Copyright 2014 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""This script grabs tweets from a redis server, and stores them in BiqQuery
using the BigQuery Streaming API.
"""
import datetime
import json
import os

import redis

import utils

# Get info on the Redis host and port from the environment variables.
# The name of this variable comes from the redis service id, 'redismaster'.
REDIS_HOST = os.environ['REDISMASTER_SERVICE_HOST']
REDIS_PORT = os.environ['REDISMASTER_SERVICE_PORT']
REDIS_LIST = os.environ['REDISLIST']

r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=0)

# Get the project ID from the environment variable set in
# the 'bigquery-controller.yaml' manifest.
PROJECT_ID = os.environ['PROJECT_ID']


def write_to_bq(bigquery):
    """Write the data to BigQuery in small chunks."""
    tweets = []
    CHUNK = 50  # The size of the BigQuery insertion batch.
    tweet = None
    mtweet = None
    count = 0
    count_max = 50000
    redis_errors = 0
    allowed_redis_errors = 3
    while count < count_max:
        while len(tweets) < CHUNK:
            # We'll use a blocking list pop -- it returns when there is
            # new data.
            res = None
            try:
                res = r.brpop(REDIS_LIST)
            except:
                print 'Problem getting data from Redis.'
                redis_errors += 1
                if redis_errors > allowed_redis_errors:
                    print "Too many redis errors: exiting."
                    return
                continue
            try:
                tweet = json.loads(res[1])
            except Exception, e:
                print e
                if redis_errors > allowed_redis_errors:
                    print "Too many redis errors: exiting."
                    return
                continue
            # First do some massaging of the raw data
            mtweet = utils.cleanup(tweet)
            # We only want to write tweets to BigQuery; we'll skip 'delete' and
            # 'limit' information.
            if 'delete' in mtweet:
                continue
            if 'limit' in mtweet:
                continue
            tweets.append(mtweet)
        # try to insert the tweets into bigquery
        response = utils.bq_data_insert(bigquery, PROJECT_ID, os.environ['BQ_DATASET'],
                             os.environ['BQ_TABLE'], tweets)
        tweets = []
        count += 1
        if count % 25 == 0:
            print ("processing count: %s of %s at %s: %s" %
                   (count, count_max, datetime.datetime.now(), response))


if __name__ == '__main__':
    print "starting write to BigQuery...."
    bigquery = utils.create_bigquery_client()
    write_to_bq(bigquery)

Starting up your Kubernetes app

After optionally building and pushing your Docker image, you can begin starting up your Kubernetes app.

In Kubernetes, pods {: target='k8s' track-type='tutorial' track-name='externalLink' track-metadata-position='body'} – rather than individual application containers – are the smallest deployable units that can be created, scheduled, and managed.

A replica set {: target='k8s' track-type='tutorial' track-name='externalLink' track-metadata-position='body'} ensures that a specified number of pod replicas are running at any one time. If there are too many replicas, it will shut down some. If there are too few, it will start more. As opposed to just creating singleton pods or even creating pods in bulk, a replica set replaces pods that are deleted or terminated for any reason, such as in the case of node failure.

A Deployment {: target='k8s' track-type='tutorial' track-name='externalLink' track-metadata-position='body'} provides declarative updates for pods and replica sets. You only need to describe the desired state in a Deployment object, and the Deployment controller will change the actual state to the desired state at a controlled rate for you.

You will use Deployments for the Redis-to-BigQuery and Twitter-to-Redis pods.

Start the Redis master pod

Start the Redis master pod first. If you're curious, you can see the specification file for the Redis master pod in kube-redis-bq/redis/redis-master.yaml:

apiVersion: v1
kind: ReplicationController
metadata:
  name: redis-master
  labels:
    name: redis-master
spec:
  replicas: 1
  template:
    metadata:
      labels:
        name: redis-master
    spec:
      containers:
      - name: master
        image: redis
        ports:
        - containerPort: 6379

1. To start the pod, run the following command:

   kubectl create -f kube-redis-bq/redis/redis-master.yaml
   

2. Run the following command to verify that the Redis master pod is running:

   kubectl get pods
   

Look at the STATUS field to verify that the master is running. In addition, you might see some other pods already running. These extra pods are added and used by the Kubernetes framework.

It can take around 30 seconds for the pod to be placed on an instance. During this period, the Redis master pod will list its host as <unassigned>. However, the final value of Host will be the name of the instance that the pod is running on. When the process completes, the pod's status will change from ContainerCreating to Running.

Start the Redis master's service

In Kubernetes, a service defines a logical set of pods and a policy by which to access them. The service is specified in kube-redis-bq/redis/redis-master-service.yaml as follows:

apiVersion: v1
kind: Service
metadata:
  name: redismaster
  labels:
    name: redis-master
spec:
  ports:
    # The port that this service should serve on.
    # You don't need to specify the targetPort if it is the same as the port,
    # though here we include it anyway, to show the syntax.
  - port: 6379
    targetPort: 6379
  selector:
    name: redis-master

The Redis pod that you created above has the label name: redis-master. The selector field of the service determines which pods will receive the traffic sent to the service.

1. Enter the following command to start the service:

   kubectl create -f kube-redis-bq/redis/redis-master-service.yaml
   

2. To view the running services, enter the following command:

   kubectl get services
   

Start the Redis-to-BigQuery Deployment

Next, start the Deployment for the pods that pull tweets from the Redis cache and stream them to your BigQuery table.

Before you start the Deployment, you might find it useful to take a closer look at kube-redis-bq/redis/bigquery-controller.yaml. This file defines the Deployment, and specifies a replica set with two pod replicas. The characteristics of these replicated pods are defined in the file by using a pod template.

1. Run the following command to start up the replicated pods:

   kubectl create -f kube-redis-bq/redis/bigquery-controller.yaml
   

2. To verify that the pods are running, run the following:

   kubectl get pods
   

   It can take about 30 seconds for your new pods to move from ContainerCreating to Running. You should see one Redis pod labeled redis-master, and two BigQuery-to-Redis pods labeled bigquery-controller.

3. You can run:

   kubectl get deployments

to see the system's defined Deployments, and how many replicas each is specified to have.

Start the Twitter-to-Redis Deployment

After your Redis master and Redis-to-BigQuery pipeline pods are running, start the Deployment that pulls in tweets and adds them to your Redis cache. Like kube-redis-bq/redis/bigquery-controller.yaml, the Twitter-to-Redis specification file, kube-redis-bq/redis/twitter-stream.yaml, defines a Deployment. However, this time the Deployment only asks for one replicated pod, as Twitter only allows one streaming API connection at a time.

1. To start up the Deployment, enter the following command:

   kubectl create -f kube-redis-bq/redis/twitter-stream.yaml
   

2. Verify that all of your pods are now running by running the following command. Again, it can take about 30 seconds for your new pod to move from ContainerCreating to Running.

   kubectl get pods
   

In addition to the pods you saw in the previous step, you should see a new pod labeled twitter-stream. Congratulations! Your analysis pipeline is now up and running.

1. You can again run:

   kubectl get deployments

to see the system's defined Deployments, and how many replicas each is specified to have. You should now see something like this:

  NAME                   DESIRED   CURRENT   UP-TO-DATE   AVAILABLE   AGE
  bigquery-controller    2         2         2            2           2m
  twitter-stream         1         1         1            1           2m

Querying your BigQuery table

Open the BigQuery web UI and click Compose Query to begin writing a new query. You can verify that it is working by running a simple query, as follows:

SELECT
  text
FROM
  [rtda.tweets]
LIMIT
  1000 IGNORE CASE;

Let your pipeline collect tweets for a while – a few hours should do, but the longer you let it run, the richer your data set will be. After you have some more data in your BigQuery table, you can try running some interesting sample queries.

This sample query demonstrates how to find the most retweeted tweets in your table, filtering on a specific term (in this case, "android"):

SELECT
  text,
  MAX(retweeted_status.retweet_count) AS max_retweets,
  retweeted_status.user.screen_name
FROM
  [rtda.tweets]
WHERE
  text CONTAINS 'android'
GROUP BY
  text,
  retweeted_status.user.screen_name
ORDER BY
  max_retweets DESC
LIMIT
  1000 IGNORE CASE;

You might also find it interesting to filter your collected tweets by a set of terms. The following query filters by the words "Kubernetes," "BigQuery," "Redis," or "Twitter:"

SELECT
  created_at,
  text,
  id,
  retweeted_status.retweet_count,
  user.screen_name
FROM
  [rtda.tweets]
WHERE
  text CONTAINS 'kubernetes'
  OR text CONTAINS 'BigQuery'
  OR text CONTAINS 'redis'
  OR text CONTAINS 'twitter'
ORDER BY
  created_at DESC
LIMIT
  1000 IGNORE CASE;

The following query looks for a correlation between the number of favorites and the number of retweets in your set of tweets:

SELECT
  CORR(retweeted_status.retweet_count, retweeted_status.favorite_count),
  lang,
COUNT(*) c
FROM [rtda.tweets]
GROUP BY lang
HAVING c > 2000000
ORDER BY 1

Going even deeper, you could also investigate whether the speakers of a specific language prefer favoriting to retweeting, or vice versa:

SELECT
  CORR(retweeted_status.retweet_count, retweeted_status.favorite_count),
  lang,
COUNT(*) c,
AVG(retweeted_status.retweet_count) avg_rt,
AVG(retweeted_status.favorite_count) avg_fv,
AVG(retweeted_status.retweet_count)/AVG(retweeted_status.favorite_count) ratio_rt_fv
FROM [rtda.tweets]
WHERE retweeted_status.retweet_count > 1 AND retweeted_status.favorite_count > 1
GROUP BY lang
HAVING c > 1000000
ORDER BY 1;