我正在关注Spark的流媒体guide。我没有使用nc -lk 9999
,而是创建了自己的简单Python服务器,如下所示。从下面的代码可以看出,它将随机生成字母a
到z
。你知道吗
import socketserver
import time
from random import choice
class AlphaTCPHandler(socketserver.BaseRequestHandler):
def handle(self):
print('AlphaTCPHandler')
alphabets = list('abcdefghikjklmnopqrstuvwxyz')
try:
while True:
s = f'{choice(alphabets)}'
b = bytes(s, 'utf-8')
self.request.sendall(b)
time.sleep(1)
except BrokenPipeError:
print('broken pipe detected')
if __name__ == '__main__':
host = '0.0.0.0'
port = 301
server = socketserver.TCPServer((host, port), AlphaTCPHandler)
print(f'server starting {host}:{port}')
server.serve_forever()
我用客户机代码测试了这个服务器,如下所示。你知道吗
import socket
import sys
import time
HOST, PORT = 'localhost', 301
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.connect((HOST, PORT))
print('socket opened')
while True:
received = str(sock.recv(1024), 'utf-8')
if len(received.strip()) > 0:
print(f'{received}')
time.sleep(1)
finally:
sock.close()
print('socket closed')
然而,我的Spark流代码似乎没有收到任何数据,或者它没有打印任何东西。代码如下。你知道吗
from pyspark.streaming import StreamingContext
from time import sleep
ssc = StreamingContext(sc, 1)
ssc.checkpoint('/tmp')
lines = ssc.socketTextStream('0.0.0.0', 301)
words = lines.flatMap(lambda s: s.split(' '))
pairs = words.map(lambda word: (word, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
counts.pprint()
ssc.start()
sleep(5)
ssc.stop(stopSparkContext=False, stopGraceFully=True)
我从输出中看到的只是下面的重复模式。你知道吗
------------------------------------------- Time: 2019-10-31 08:38:22 ------------------------------------------- ------------------------------------------- Time: 2019-10-31 08:38:23 ------------------------------------------- ------------------------------------------- Time: 2019-10-31 08:38:24 -------------------------------------------
你知道我做错了什么吗?你知道吗
你的流代码工作正常。是你的服务器给它提供了错误的信息-每个字母后面没有行分隔符,所以Spark看到的是一个不断增长的行,它只是一直在等待该行完成,这从来没有发生过。修改服务器以在每个字母中发送新行:
结果是:
相关问题 更多 >
编程相关推荐