import json

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from sentiment_nltk import classify_sentiment

sc = SparkContext(appName="PythonSparkStreamingKafka")
sc.setLogLevel("WARN")

ssc = StreamingContext(sc, 10)
kafkaStream = KafkaUtils.createStream(ssc, 'kafka:2181', 'sentiment-analyser', {'brexit': 1})


def processTweet(tweet):
    tweet_text = json.loads(tweet)["text"]
    #print(tweet_text)

    (sentiment, confidence) = classify_sentiment(tweet_text, False)

    print("----------------\n%s\n>>>>>> %s (%f)" % (tweet_text, sentiment, confidence))


kafkaStream.foreachRDD(lambda rdd: rdd.foreach(lambda row: processTweet(row[1])))

# start the Spark streaming
ssc.start()
ssc.awaitTermination()

cf3c04226e0318dadf5628306820d8387db23608
<mask token>


def processTweet(tweet):
    tweet_text = json.loads(tweet)['text']
    sentiment, confidence = classify_sentiment(tweet_text, False)
    print('----------------\n%s\n>>>>>> %s (%f)' % (tweet_text, sentiment,
        confidence))


<mask token>

<mask token>
sc.setLogLevel('WARN')
<mask token>


def processTweet(tweet):
    tweet_text = json.loads(tweet)['text']
    sentiment, confidence = classify_sentiment(tweet_text, False)
    print('----------------\n%s\n>>>>>> %s (%f)' % (tweet_text, sentiment,
        confidence))


kafkaStream.foreachRDD(lambda rdd: rdd.foreach(lambda row: processTweet(row
    [1])))
ssc.start()
ssc.awaitTermination()

<mask token>
sc = SparkContext(appName='PythonSparkStreamingKafka')
sc.setLogLevel('WARN')
ssc = StreamingContext(sc, 10)
kafkaStream = KafkaUtils.createStream(ssc, 'kafka:2181',
    'sentiment-analyser', {'brexit': 1})


def processTweet(tweet):
    tweet_text = json.loads(tweet)['text']
    sentiment, confidence = classify_sentiment(tweet_text, False)
    print('----------------\n%s\n>>>>>> %s (%f)' % (tweet_text, sentiment,
        confidence))


kafkaStream.foreachRDD(lambda rdd: rdd.foreach(lambda row: processTweet(row
    [1])))
ssc.start()
ssc.awaitTermination()

import json
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from sentiment_nltk import classify_sentiment
sc = SparkContext(appName='PythonSparkStreamingKafka')
sc.setLogLevel('WARN')
ssc = StreamingContext(sc, 10)
kafkaStream = KafkaUtils.createStream(ssc, 'kafka:2181',
    'sentiment-analyser', {'brexit': 1})


def processTweet(tweet):
    tweet_text = json.loads(tweet)['text']
    sentiment, confidence = classify_sentiment(tweet_text, False)
    print('----------------\n%s\n>>>>>> %s (%f)' % (tweet_text, sentiment,
        confidence))


kafkaStream.foreachRDD(lambda rdd: rdd.foreach(lambda row: processTweet(row
    [1])))
ssc.start()
ssc.awaitTermination()

import json

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from sentiment_nltk import classify_sentiment

sc = SparkContext(appName="PythonSparkStreamingKafka")
sc.setLogLevel("WARN")

ssc = StreamingContext(sc, 10)
kafkaStream = KafkaUtils.createStream(ssc, 'kafka:2181', 'sentiment-analyser', {'brexit': 1})


def processTweet(tweet):
    tweet_text = json.loads(tweet)["text"]
    #print(tweet_text)

    (sentiment, confidence) = classify_sentiment(tweet_text, False)

    print("----------------\n%s\n>>>>>> %s (%f)" % (tweet_text, sentiment, confidence))


kafkaStream.foreachRDD(lambda rdd: rdd.foreach(lambda row: processTweet(row[1])))

# start the Spark streaming
ssc.start()
ssc.awaitTermination()
