Spark Streaming连接被拒绝排查：Colab外部数据接入故障

运行代码 #

修改后的Spark Streaming官方示例network_wordcount.py代码：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __name__ == "__main__":
    sc = SparkContext(appName="PythonStreamingNetworkWordCount")
    ssc = StreamingContext(sc, 1)                                  # 第二个参数为批次间隔（秒）

    lines = ssc.socketTextStream('localhost', 5101)
    counts = lines.flatMap(lambda line: line.split(" "))\
                  .map(lambda word: (word, 1))\
                  .reduceByKey(lambda a, b: a + b)
    
    counts.pprint()
    #counts.toDF().write.csv("outputDir",mode="append")
    #counts.saveAsTextFiles('output','pm')
    
    ssc.start()
    ssc.awaitTermination()

部署操作步骤 #

在Google Colab中执行以下步骤：

• 创建ngrok TCP隧道，将本地1234端口暴露至公网：

  public_url = ngrok.connect(1234,"tcp")
  

• 在colab-xterm中用socat实现1234端口到5101端口的转发：

  socat TCP4-LISTEN:1234 TCP4:localhost:5101
  

• 在第二个xterm中提交Spark任务：

  spark-submit wcstream.py
  

• 程序启动后，通过Colab单元格向ngrok公网地址（示例：tcp://0.tcp.ngrok.io:17078）推送数据：

  !echo 'Hello throuugh TCP tunnel 4' > /dev/tcp/0.tcp.ngrok.io/17078
  

错误现象 #

执行数据推送操作后，socat抛出连接拒绝错误：

/content# socat TCP4-LISTEN:1234 TCP4:localhost:5101
2023/05/30 05:25:49 socat[8041] E connect(5, AF=2 127.0.0.1:5101, 16): Connection refused

已知用netcat本地推送数据可正常运行，但需要实现外部数据源向Colab中的Spark Streaming推送数据，需排查问题并解决。

排查原因与解决方法 #

核心问题：启动顺序错误 #

socat TCP4-LISTEN:1234 TCP4:localhost:5101的逻辑是：当外部请求连接1234端口时，socat会主动连接localhost:5101。若先启动socat再启动Spark任务，socat尝试连接5101时，Spark还未启动监听该端口，就会触发Connection refused错误。

解决步骤 #

1. 调整启动顺序：
   • 先提交Spark任务，等待Spark启动并监听5101端口（可通过netstat -tulpn | grep 5101确认端口处于监听状态）。
   • 再启动socat端口转发。
   • 最后执行数据推送操作，此时socat能正常连接到Spark监听的5101端口，数据可正常转发。

优化方案：省略socat直接转发 #

让Spark直接监听ngrok暴露的1234端口，省去socat转发步骤：

1. 修改代码中的端口配置，将ssc.socketTextStream('localhost', 5101)改为ssc.socketTextStream('localhost', 1234)。
2. 创建ngrok TCP隧道：public_url = ngrok.connect(1234,"tcp")。
3. 提交Spark任务：spark-submit wcstream.py。
4. 直接向ngrok公网地址推送数据，无需socat转发，数据可直接到达Spark的监听端口。

额外验证建议 #

• 启动Spark后，在Colab中执行!nc -zv localhost 5101（或1234，根据配置），确认端口已处于监听状态，再进行后续操作。
• 若使用socat，可添加-d -d参数查看详细日志定位问题：socat -d -d TCP4-LISTEN:1234 TCP4:localhost:5101

内容的提问来源于stack exchange，提问作者Calcutta