<jupyter_start><jupyter_text>**General Imports**<jupyter_code>from operator import add
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from requests_oauthlib import OAuth1Session
from time import gmtime, strftime

import ast
import json
import nltk
import requests
import requests_oauthlib
import string
import time<jupyter_output><empty_output><jupyter_text>**NLTK Imports**<jupyter_code>from nltk.classify import NaiveBayesClassifier
from nltk.corpus import stopwords
from nltk.corpus import subjectivity
from nltk.sentiment import SentimentAnalyzer
from nltk.sentiment.util import *

import nltk
BATCH_INTERVAL = 5
ssc = StreamingContext(sc, BATCH_INTERVAL)<jupyter_output><empty_output><jupyter_text>Training the Sentiment Analysis Classifier An essential part of creating a sentiment analysis algorithm (or any data mining algorithm) is to use a comprehensive dataset or "Corpus" for learning, as well as a set of test data to ensure that the accuracy of the your algorithm meets the standards you expect. This will also allow you to adjust your algorithm in order to predict better (or more precisely) natural language features that you could extract from the text that will contribute to the sentiment classification, rather than using a generic approach. We will use the training dataset provided by the University of Michigan for Kaggle competitions: https://inclass.kaggle.com/c/si650winter11 This dataset contains 1,578,627 classified tweets and each row is marked as:- 1 for positive feeling- 0 for negative feeling<jupyter_code>file = sc.textFile('aux/datasets/sentiment-analysis.csv')
header = file.take(1)[0]
dataset = file.filter(lambda line: line != header)
dataset.take(3)
def clearLine(line):
    columns = line.split(',')
    translator = str.maketrans({key: None for key in string.punctuation})
    
    sentiment = columns[1]
    tweet = columns[3].strip()
    tweet = tweet.translate(translator)
    tweet = tweet.split(' ')
    
    tweetLower = [word.lower() for word in tweet]
    
    return(tweetLower, sentiment)
datasetTraining = dataset.map(lambda line: clearLine(line))
datasetTraining.take(3)
nltk.download('stopwords')
stopWords = [[word, word + '_NEG'] for word in stopwords.words('english') for singleWord in word]
stopWords = [word for wordPair in stopWords for word in wordPair]
sentimentAnalyzer = SentimentAnalyzer()
datasetTrainingSample = datasetTraining.take(10000)
wordsNegative = sentimentAnalyzer.all_words([mark_negation(word) for word in datasetTrainingSample])
wordsNegative = [word for word in wordsNegative if word not in stopWords]
unigramFeatures = sentimentAnalyzer.unigram_word_feats(wordsNegative, top_n = 200)
sentimentAnalyzer.add_feat_extractor(extract_unigram_feats, unigrams = unigramFeatures)
trainingSet = sentimentAnalyzer.apply_features(datasetTrainingSample)<jupyter_output><empty_output><jupyter_text>Model Training<jupyter_code>trainer = NaiveBayesClassifier.train
classifier = sentimentAnalyzer.train(trainer, trainingSet)
testSentence01 = [(['this', 'program', 'is', 'bad'], '')]
testSentence02 = [(['tough', 'day', 'at', 'work', 'today'], '')]
testSentence03 = [(['that', 'place', 'is', 'awesome'], '')]
testSet01 = sentimentAnalyzer.apply_features(testSentence01)
testSet02 = sentimentAnalyzer.apply_features(testSentence02)
testSet03 = sentimentAnalyzer.apply_features(testSentence03)<jupyter_output><empty_output><jupyter_text>**Twitter Authentication**<jupyter_code>apiKey = ''
apiKeySecret = ''
accessToken = ''
accessTokenSecret = ''
searchTerm = 'Trump'
urlSample = 'https://stream.twitter.com/1.1/statuses/sample.json'
urlFilter = 'https://stream.twitter.com/1.1/statuses/filter.json?track=' + searchTerm
auth = requests_oauthlib.OAuth1(apiKey, apiKeySecret, accessToken, accessTokenSecret)<jupyter_output><empty_output><jupyter_text>**Stream Configuration**<jupyter_code>rdd = ssc.sparkContext.parallelize([0])
stream = ssc.queueStream([], default = rdd)
NUM_TWEETS = 500
def twitterDataStream():
    response = requests.get(urlFilter, auth = auth, stream = True)
    print(urlFilter, response)
    
    count = 0
    
    for line in response.iter_lines():
        try:
            if count > NUM_TWEETS:
                break

            post = json.loads(line.decode('utf-8'))
            contents = [post['text']]
            count += 1

            yield str(contents)
        except:
            result = False
def mapData(t, rdd):
    return rdd.flatMap(lambda x: twitterDataStream())
stream = stream.transform(mapData)
streamCoord = stream.map(lambda line: ast.literal_eval(line))
def tweetClassifier(tweet):
    sentence = [(tweet, '')]
    testSet = sentimentAnalyzer.apply_features(sentence)
    
    print(tweet, classifier.classify(testSet[0][0]))
    
    return(tweet, classifier.classify(testSet[0][0]))
def getTweetText(rdd):
    for line in rdd:
        tweet = line.strip()
        translator = str.maketrans({key: None for key in string.punctuation})
        tweet = tweet.translate(translator)
        tweet = tweet.split(' ')
        
        tweetLower = [word.lower() for word in tweet]
        
        return tweetClassifier(tweetLower)
results = []
def rddOutput(rdd):
    global results
    
    pairs = rdd.map(lambda x: (getTweetText(x)[1], 1))
    counts = pairs.reduceByKey(add)
    output = [count for count in counts.collect()]
    
    result = [time.strftime('%I:%M:%S'), output]
    results.append(result)
    
    print(result)    
streamCoord.foreachRDD(lambda t, rdd: rddOutput(rdd))<jupyter_output><empty_output><jupyter_text>Start Streaming<jupyter_code>ssc.start()
count = True
while count:
    if len(results) > 5:
        count = False
outputFile = 'aux/' + time.strftime('%I%M%S')
rddResults = sc.parallelize(results)
rddResults.saveAsTextFile(outputFile)
rddResults.collect()
ssc.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1065350578624915, 'fraction_numerical': 0.011402314499659632, 'mean_word_length': 4.960446247464503, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '248278', 'license': 'MIT', 'n_tokens_mistral': 1888, 'n_tokens_neox': 1802, 'n_words': 460, 'path': 'modules/08-real-time-twitter-sentiment-analysis-with-spark-streaming-and-nltk/02-real-time-twitter-sentiment-analysis.ipynb', 'repo_name': 'cfascina/rtaps', 'size': '5876'}