Почему нельзя запустить несколько SparkContexts одновременно - PullRequest
Новая
       26

Почему нельзя запустить несколько SparkContexts одновременно

0 голосов
/ 08 декабря 2018

я использую python 2 и spark.Я следую инструкции по подсчету слов в твиттере по этой ссылке https://github.com/Ruthvicp/CS5590_BigDataProgramming/wiki/Lab-Assignment-4----Spark-MLlib-classification-algorithms,-word-count-on-twitter-streaming У меня есть 2 файла TSWordCount 

import findspark
findspark.init()
import pyspark
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql.functions import desc
from collections import namedtuple
import os
os.environ["SPARK_HOME"] = "C:\\spark-2.3.1-bin-hadoop2.7\\spark-2.3.1-bin-hadoop2.7"
os.environ["HADOOP_HOME"] = "C:\\winutils\\"

def main():
    sc =SparkContext(appName="Countwords1234")
    wordcount = {}
    ssc = StreamingContext(sc, 5)
    lines = ssc.socketTextStream("localhost", 5678)
    fields = ("word", "count")
    Tweet = namedtuple('Text', fields)
    # lines = socket_stream.window(20)
    counts = lines.flatMap(lambda text: text.split(" "))\
        .map(lambda x: (x, 1))\
        .reduceByKey(lambda a, b: a + b).map(lambda rec: Tweet(rec[0], rec[1]))
    counts.pprint()
    ssc.start()
    ssc.awaitTermination()
if __name__ == "__main__":
    main()

Когда я запускаю этот файл, он успешен, и вывод «Прослушивание порта 5678»"и мой второй файл - TwitterListener 

import findspark
findspark.init()
import pyspark
import tweepy
from tweepy import OAuthHandler
from tweepy import Stream
from tweepy.streaming import StreamListener
import socket
import json
import time

consumer_key = '30f****'
consumer_secret = 'smu7B******
access_token = '153*******'
access_secret = 'QIizsB***'


auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_secret)

class TweetsListener(StreamListener):

    def __init__(self, csocket):
        self.client_socket = csocket

    def on_data(self, data):
        try:
            msg = json.loads(data)
            print(msg['text'].encode('utf-8'))
            self.client_socket.send(msg['text'].encode('utf-8'))
            return True
        except BaseException as e:
            print("Error on_data: %s" % str(e))
        return True

    def on_error(self, status):
        print(status)
        return True


def sendData(c_socket):
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_secret)

    twitter_stream = Stream(auth, TweetsListener(c_socket))
    twitter_stream.filter(track=['fifa'])


if __name__ == "__main__":
    s = socket.socket()  # Create a socket object
    host = "localhost"  # Get local machine name
    port = 5678 # Reserve a port for your service.
    s.bind((host, port))  # Bind to the port
    print("Listening on port: %s" % str(port))
    s.listen(5)  # Now wait for client connection.
    c, addr = s.accept()  # Establish connection with client.
    print("Received request from: " + str(addr))
    time.sleep(5)
    sendData(c)

Как вы видите, файл слушателя в Твиттере слушает порт localhost: 5678.Затем в файле TSWordCount я использую SparkContext (appname = ""), я думаю, что я должен поместить здесь название своего приложения в твиттер, поэтому я поместил Countwors124 там.Затем я вызываю порт по ssc.socketTextStream ("localhost", 5678).Но у меня есть ошибка в TSWordCount, когда я запускаю, появляется ошибка, которая говорит о том, что не может запустить несколько SparkContexts одновременно;существующий SparkContext (app = PySparkShell, master = local [*]), созданный в результате поиска ошибки, и я нашел решение, подобное использованию sc.stop (), так что я поставил его после ssc.awaitTermination ().Но это не сработало.Что мне теперь делать ?

1 Ответ

0 голосов
/ 09 декабря 2018

Я нашел ответ.я заменил sc =SparkContext(appName="Countwords1234") на sc = SparkContext.getOrCreate() и все заработало.хотя я все еще не понимаю, в конце дня результат имеет значение LOL

...