Streaming tweets with Twitter API

Get Twitter API account

This was suggested preparations before the course started: ...head over to the Twitter Developer page and apply for a Developer Account. Get "Essential" access first. Then you can go on to apply for wider "Academic Research" access (but: that is mainly for Master students who have started the thesis work, so I am not sure what Twitter will say...). It is not mandatory to use Twitter data in the course, but it will be a convenient way to get started.

The simple Essential account is a fine start and offers a lot of opportunities. When you have defined a more concrete project, you can apply for an Academic Research account that gives even wider access. Then it is up to Twitter whether they will grant you access.

Usually, a project that gives Essential access will be granted quickly. Twitter offers several ways to authenticate yourself, but the easiest was is to use the Bearer Token. You can find it from your Twitter project page, under Keys and tokens. In your info319-exercises folder, create a keys/bearer_token file that only you can read:

(venv) $ mkdir keys
(venv) $ chmod 700 keys
(venv) $ touch keys/bearer_token
(venv) $ chmod 600 keys/bearer_token
(venv) $ echo "...the long alpha-numeric token string goes here..." > keys/bearer_token

If you use git, you can add the line /keys/ to your .gitignore file as well. You do not want it uploaded to an external repository.

Saving tweets to file

Install the tweepy API:

(venv) $ pip install tweepy

Task 1: Run the following minimal boilerplate code to connect to Twitter's API and start downloading a stream of tweets (the code takes a few seconds to connect):

import json
import time
import tweepy

KEY_FILE = './keys/bearer_token'
DURATION = 10  # how many seconds to run for, including connection time
   
def on_data(json_data):
    """Tweepy calls this when it receives data from Twitter"""
    json_obj = json.loads(json_data.decode())
    print('Received tweet:', json_obj)

# Initiate the tweepy client and start the sampling thread.
bearer_token = open(KEY_FILE).read().strip()
streaming_client = tweepy.StreamingClient(bearer_token)
streaming_client.on_data = on_data
streaming_client.sample(threaded=True)
time.sleep(DURATION)
streaming_client.disconnect()

Note how, every time a new tweet is received, tweepy calls the on_data()-function, which prints the tweeted text to the screen. This is just a simple example to get started. In Session 1 we will look at many other ways to access the Twitter API V2 through tweepy.

Task 2: Change the code to save the tweets to a file (for example as JSON or CSV). Create a subfolder for tweet files to keep your info319-exercise2 folder tidy.

Task 3: Change the code so you download and save additional information about each tweet, for example the handle (user name) of the tweeter, whether it is a retweet of another id, and perhaps the date and time.

Task 4: Refactor the code to prepare for cleaner termination by creating a new on_finish()-function, which disconnects the streaming client, closes open files, and performs any other needed termination tasks. Remove time.sleep(DURATION), and instead make the on_data()-function call on_finish() when a given number of tweets (say 100 to start) has been received.

Streaming tweets to a socket

Task 5: Change the code to write the stream of tweets to a network socket instead. Think of a socket as an internal internet connection from your machine back to itself. Different programs on your computer can use this socket to communicate using the regular internet APIs.

Here are some critical pieces of code you need to include in your program:

import socket

HOST = 'localhost'
PORT = 65000  # for example
NUM_TWEETS = 100

tweet_count = 0

To open a socket for output (as server side):

socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.bind((HOST, PORT))
socket.listen()
connection, address = socket.accept()

To send data on the socket:

def on_data(json_data):
    json_obj = json.loads(json_data.decode())
    json_bytes = json.dumps(json_obj['data']).encode()
    connection.sendall(json_bytes)
    if tweet_count >= NUM_TWEETS:
        ...close the socket etc...

The code is based on this example.

You can use the nc utility to receive data from the socket. From another console window (inside or outside VS Code, a virtual environment is not needed):

$ apt install nc
$ nc localhost 65000  # you must start the tweet stream before you run this command

IMPORTANT: When you debug, the socket will often remain open after your program has crashed, or if you are in the same Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).

Stream to Spark

Streaming Spark is the topic of Session 3, but you can run the following script spark_receiver.py to receive tweets from the socket as a Spark stream:

import os
import findspark
from pyspark.sql import SparkSession

HOST = os.environ['HOST'] if 'HOST' in os.environ else 'localhost'
PORT = os.environ['PORT'] if 'PORT' in os.environ else '65000'
PORT = int(PORT)

findspark.init()
spark = SparkSession.builder.appName('SparkTest').getOrCreate()
spark.sparkContext.setLogLevel('ERROR')

# incoming stream
incoming_df = spark.readStream \
                .format('socket') \
                .option('host', HOST) \
                .option('port', PORT) \
                .load()
# outgoing stream
outgoing_df = incoming_df.select('value').writeStream \
                .format('console') \
                .outputMode('update') \
                .start() \
                .awaitTermination()

The code is based on this example:

Start the streaming tweet harvester first and then start the script with, e.g.:

(venv) $ export HOST=localhost PORT=65000; python spark_receiver.py

Instead of using import findspark; findspark.init() you can set the $SPARK_HOME environment variable, for example:

export SPARK_HOME=${PWD}/venv/lib/python3.8/site-packages/pyspark

Add this line to the end of your ~/.bashrc file if you want to make it permanent. It is safer than findspark when you have several Spark installations on the same machine.