Отправить данные Spark Streaming обратно клиенту - PullRequest
Новая
       42

Отправить данные Spark Streaming обратно клиенту

0 голосов
/ 27 апреля 2020

Я новичок в Apache Spark Streaming. Я разрабатываю приложение для потоковой передачи с искрой, чтобы найти кратчайший путь, и снова отправляю его обратно клиенту. Я написал код для сбора и обработки данных с помощью функции, но у меня есть проблема, как я могу снова отправить результаты обратно клиенту Вот мой код: 

import networkx as nx
from pyspark import SparkConf,SparkContext
from pyspark.streaming import StreamingContext
TCP_IP = "127.0.0.1"
TCP_PORT = 5000

# Creating a Spark Configuration
conf=SparkConf()
conf.setAppName('ShortestPathApp')

sc= SparkContext(conf)
ssc= StreamingContext(sc,2)

def shortestPath(line):
    # get the values from rdd
    vehicleId = line[0]
    source = line[1]
    destination = line[2]
    deadline = line[3]

    # find shortest path
    shortest = nx.dijkstra_path(G, source, destination)



# receive from Socket
dataStream =ssc.socketTextStream(TCP_IP,TCP_PORT)
vehicle_data = dataStream.map(lambda line: line.split(" "))
vehicle_data.foreachRDD(lambda rdd: rdd.foreach(shortestPath))
ssc.start()
ssc.awaitTermination()

Как я могу отправить данные обратно клиенту

1 Ответ

2 голосов
/ 27 апреля 2020

Используйте StreamingContext для вывода sh данных обратно к месту назначения в виде потока. Вы можете создать его, как показано ниже. Добавить метод для получения одноэлементного экземпляра SparkSession 

# Lazily instantiated global instance of SparkSession
def getSparkSessionInstance(sparkConf):
    if ("sparkSessionSingletonInstance" not in globals()):
        globals()["sparkSessionSingletonInstance"] = SparkSession \
            .builder \
            .config(conf=sparkConf) \
            .getOrCreate()
    return globals()["sparkSessionSingletonInstance"]


sparkSess = getSparkSessionInstance(rdd.context.getConf()) 
vehicle_data_df = sparkSess.createDataFrame(vehicle_data)
vehicle_data_df.writeStream\
      .format("socket")\
      .option("host",TCP_OUTPUT_IP)    //Output socket IP address 
      .option("port",TCP_OUTPUT_PORT)  //Output socket port 
      .outputMode('append')\
      .start()\
      .awaitTermination()
...