Tweepy Streaming Socket не может отправить предварительно обработанный текст - PullRequest
0 голосов
/ 18 октября 2019

У меня есть две программы, которые подключаются через сокеты. Одним из них является tweepy StreamListener, где я также предварительно обрабатываю данные с помощью библиотеки «tweet-preprocessor». Другая программа должна подключиться к этому сокету и проанализировать данные с помощью Spark Structured Streaming. Проблема в том, что Spark не получает пакеты при предварительной обработке данных перед их отправкой.

Это StreamListener

import tweepy
import socket
import json
import preprocessor as p

CONSUMER_KEY = ""
CONSUMER_SECRET = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

p.set_options(p.OPT.URL, p.OPT.EMOJI, p.OPT.SMILEY)

class MyStreamListener(tweepy.StreamListener):
    def __init__(self, csocket):
        self.client_socket = csocket

    def on_data(self, raw_data):
        try:
            data = json.loads(raw_data)
            clean_text = p.clean(data["text"])
            print(clean_text)
            self.client_socket.send(clean_text.encode("utf-8"))
            return True
        except BaseException as e:
            print("Error: " + str(e))
        return True

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


skt = socket.socket()
host = "localhost"
port = 5555
skt.bind((host, port))
skt.listen()
client, address = skt.accept()

myStreamListener = MyStreamListener(csocket=client)
myStream = tweepy.Stream(auth=auth, listener=myStreamListener, )
myStream.filter(track=["Trump"], languages=["en"])

И простой код Spark:

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, size

spark = SparkSession.builder.appName("TwitterSpark").getOrCreate()

lines = spark.readStream.format("socket").option("host", "localhost").option("port", 5555).load()

#tweetlength = lines.select(
#        size(split(lines.value, " ")).alias("tweetlength")
#)

query = lines.writeStream.outputMode("update").format("console").start()

query.awaitTermination()

1 Ответ

0 голосов
/ 18 октября 2019

Скорее всего, clean_text не имеет символа новой строки (\n) в конце. В отличие от print(clean_text), который автоматически добавляет новую строку, socket.send() отправляет байты из clean_text.encode("utf-8") как есть, и вам нужно явно добавить \n:

self.client_socket.send((clean_text + "\n").encode("utf-8"))

Без \nчтобы разделить строки в данных сокета, Spark рассматривает входные данные как одну растущую строку, если только в самом тексте твита нет новых строк.

...