Lab_3_Jupyter_Notebook_1_Twitter_to_Kafka.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# B5 Big Data Tools and Technologies - Lab 3-1: How to Publish Tweets to a Kafka Topic"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook consists of python code that can be run from an SSH session but it is placed here for convenience."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Import libraries required to communicate with the twitter API from python\n",
    "import tweepy\n",
    "from tweepy import OAuthHandler\n",
    "\n",
    "# Enter your credentials below\n",
    "consumer_key = 'n99VMDRF7VbR1oRo2LKkkkI9I'\n",
    "consumer_secret = '5cqRaWXbL6Mzr7R8r9PxS4WleVjQxEYb9rGUYM59cLeYmn0O0i'\n",
    "access_token = '797344-lJOKwqXmw3tz7iyODKir7Dq2bUV8Kyz7atWqgswlx7C'\n",
    "access_secret = '4X2zQlZKEtNrgzxaLpeqMGhgfjgQf3JabLcpI3W09TY2q'\n",
    "\n",
    "# Connect to twitter API using the credentials provided\n",
    "auth = OAuthHandler(consumer_key, consumer_secret)\n",
    "auth.set_access_token(access_token, access_secret)\n",
    "\n",
    "api = tweepy.API(auth)\n",
    "\n",
    "# Import libraries required to run Kafka in python\n",
    "from kafka import KafkaConsumer, KafkaProducer\n",
    "import json\n",
    "\n",
    "producer = KafkaProducer(bootstrap_servers='localhost:9092')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Send a test message to the Kafka topic \"twitterstream\". You should see the test message in the console or in the output file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<kafka.producer.future.FutureRecordMetadata at 0x7f18d870c1d0>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "producer.send ('twitterstream',json.dumps('test message').encode('utf-8'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Start streaming Twitter feeds. Adjust hashtags as needed in the following line:\n",
    "<pre>twitter_stream.filter(track=['#bigdata'])</pre>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from tweepy import Stream\n",
    "from tweepy.streaming import StreamListener\n",
    " \n",
    "class MyListener(StreamListener):\n",
    " \n",
    "    def on_data(self, data):\n",
    "        try:\n",
    "            with open('python.json', 'a') as f:\n",
    "                f.write(data)\n",
    "                producer.send ('twitterstream',json.dumps(data).encode('utf-8'))\n",
    "                return True\n",
    "        except BaseException as e:\n",
    "            #print(&quot;Error on_data: %s&quot; % str(e))\n",
    "            return True\n",
    " \n",
    "    def on_error(self, status):\n",
    "        \n",
    "        print(status)\n",
    "        return True\n",
    "    \n",
    "try: \n",
    "    twitter_stream = Stream(auth, MyListener())\n",
    "    twitter_stream.filter(track=['#bigdata'])\n",
    "except KeyboardInterrupt:\n",
    "    print(\"Streaming Feed Ended\")\n",
    "    "
   ]
  }
 ],
 "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.4.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}