Ingesting realtime tweets using Apache Kafka, Tweepy and Python.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ingesting realtime tweets using Apache Kafka, Tweepy and Python\n",
    "\n",
    "### Purpose:\n",
    "- main data source for the lambda architecture pipeline\n",
    "- uses twitter streaming API to simulate new events coming in every minute\n",
    "- Kafka Producer sends the tweets as records to the Kafka Broker\n",
    "\n",
    "### Contents: \n",
    "- [Twitter setup](#1)\n",
    "- [Defining the Kafka producer](#2)\n",
    "- [Producing and sending records to the Kafka Broker](#3)\n",
    "- [Deployment](#4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tweepy\n",
    "import time\n",
    "from kafka import KafkaConsumer, KafkaProducer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"1\"></a>\n",
    "### Twitter setup\n",
    "- getting the API object using authorization information\n",
    "- you can find more details on how to get the authorization here:\n",
    "https://developer.twitter.com/en/docs/basics/authentication/overview"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# twitter setup\n",
    "consumer_key = \"1\"\n",
    "consumer_secret = \"2\"\n",
    "access_token = \"3\"\n",
    "access_token_secret = \"4\"\n",
    "# Creating the authentication object\n",
    "auth = tweepy.OAuthHandler(consumer_key, consumer_secret)\n",
    "# Setting your access token and secret\n",
    "auth.set_access_token(access_token, access_token_secret)\n",
    "# Creating the API object by passing in auth information\n",
    "api = tweepy.API(auth) \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A helper function to normalize the time a tweet was created with the time of our system"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime, timedelta\n",
    "\n",
    "def normalize_timestamp(time):\n",
    "    mytime = datetime.strptime(time, \"%Y-%m-%d %H:%M:%S\")\n",
    "    mytime += timedelta(hours=1)   # the tweets are timestamped in GMT timezone, while I am in +1 timezone\n",
    "    return (mytime.strftime(\"%Y-%m-%d %H:%M:%S\")) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"2\"></a>\n",
    "### Defining the Kafka producer\n",
    "- specify the Kafka Broker\n",
    "- specify the topic name\n",
    "- optional: specify partitioning strategy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "producer = KafkaProducer(bootstrap_servers='localhost:9092')\n",
    "topic_name = 'tweets-lambda1'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"3\"></a>\n",
    "### Producing and sending records to the Kafka Broker\n",
    "- querying the Twitter API Object\n",
    "- extracting relevant information from the response\n",
    "- formatting and sending the data to proper topic on the Kafka Broker\n",
    "- resulting tweets have following attributes:\n",
    "    - id \n",
    "    - created_at\n",
    "    - followers_count\n",
    "    - location\n",
    "    - favorite_count\n",
    "    - retweet_count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_twitter_data():\n",
    "    res = api.search(\"Apple OR iphone OR iPhone\")\n",
    "    for i in res:\n",
    "        record = ''\n",
    "        record += str(i.user.id_str)\n",
    "        record += ';'\n",
    "        record += str(normalize_timestamp(str(i.created_at)))\n",
    "        record += ';'\n",
    "        record += str(i.user.followers_count)\n",
    "        record += ';'\n",
    "        record += str(i.user.location)\n",
    "        record += ';'\n",
    "        record += str(i.favorite_count)\n",
    "        record += ';'\n",
    "        record += str(i.retweet_count)\n",
    "        record += ';'\n",
    "        producer.send(topic_name, str.encode(record))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_twitter_data()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"4\"></a>\n",
    "### Deployment \n",
    "- perform the task every couple of minutes and wait in between"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def periodic_work(interval):\n",
    "    while True:\n",
    "        get_twitter_data()\n",
    "        #interval should be an integer, the number of seconds to wait\n",
    "        time.sleep(interval)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "periodic_work(60 * 0.1)  # get data every couple of minutes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}