Python - Stream - 如何更快地将数据发送到服务器

Question

我想将文本文件中的字符串发送到我的本地端口，但我必须为每个字符串打开连接并关闭它。因此，数据流很慢。（几乎在两秒内1个字符串）。我怎样才能让它更快？

        while 1:
            conn, addr = s.accept()

            line_number = random.randint(1,2261074)
            liste.append(line_number)
            line = linecache.getline(filename,line_number)

            sendit = line.split("   ")[1]
            print type(sendit)
            print "sending: " + sendit
            conn.send(sendit)

            conn.close()
            print('End Of Stream.')

Answer 1

answer 建议在每个连接上向 Spark 发送 10 条消息，将每条消息间隔 1 秒，然后关闭连接。

保持连接打开并在服务器端使用非阻塞套接字可能会更好。

下面的服务器代码保持连接打开，在非阻塞套接字上分批发送消息，每批之间有一个空闲延迟。

这可以用来测试 Spark 接收消息的速度。我已将其设置为分批发送 50 条消息，然后等待 1 秒再发送下一个 50 条。

即使我将空闲延迟设置为零，Spark 也可以在我的机器上接收到所有消息。

您可以根据应用的需要进行试验和调整。

服务器代码：

import socket
import time
import select

def do_send(sock, msg, timeout):
    readers = []
    writers = [sock]
    excepts = [sock]
    rxs, txs, exs = select.select(readers, writers, excepts, timeout)
    if sock in exs:
        return False
    elif sock in txs:
        sock.send(msg)
        return True
    else:
        return False

host = 'localhost'
port = 9999

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((host, port))
s.listen(1)

print "waiting for client"
conn, addr = s.accept()
print "client connected"
conn.setblocking(0)

batchSize = 50
idle = 1.
count = 0
running = 1
while running:
    try:
        sc = 0
        while (sc < batchSize):
            if do_send (conn, 'Hello World ' + repr(count) + '\n', 1.0):
                sc += 1
                count += 1
        print "sent " + repr(batchSize) + ", waiting " + repr(idle) + " seconds"
        time.sleep(idle)
    except socket.error:
        conn.close()
        running = 0

print "done"

简单的 Spark 代码：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]","test")
ssc = StreamingContext(sc, 1)
ststr = ssc.socketTextStream("localhost", 9999)
lines = ststr.flatMap(lambda line: line.split('\n'))
lines.pprint()
ssc.start()  
ssc.awaitTermination()

希望这可能有用。