在Flink/Kafka中从同一个插座读写

Question

美好的一天

我正在通过flink/kafka接收数据（流数据）。我连接的端口与我需要写回消息的端口相同

TCP/IP-

// 1. Connect to TCP Stream (TCP (Socket) -> Kafka Stream INPUT)
val consumer_stream = env.socketTextStream(url, port, '\n')

// 2. Processing Data
.....

// 3. Write result to kafka topic 
consumer_stream.addSink(new FlinkKafkaProducer09[String](broker_url, topic_name, new SimpleStringSchema()))

// 4. Send result back to connected url ie.(Ref Step 1 URL) (url+port)
(This is where I need Assistance)

连接到 URL 和端口工作正常。我接收并处理写入主题的数据 现在我还需要写回我连接到的同一URL和端口。{由于 Url 和端口可以同时发送和接收数据}

我把它写到另一个端口

// write to Different PORT
val socket_write: DataStreamSink[String] = out_data.writeToSocket(url, diff_port, new SimpleStringSchema())

这个管用...问题是试图写入同一个端口。当我使用我从中读取的同一个端口时...flink作业失败

有什么想法吗

问候

Answer 1

您可以使用自定义的SinkFunction将数据写回URL。

stream.addSink(new SinkFunction<String>() {
    // initialise the client to send the data
    public void invoke(String value) throws Exception {
        // send here.               
    }
}

或使用＜code＞SocketClientSink＜/code＞进行

env.socketTextStream("localhost", 5555).map(x => { println(x); x }).addSink(new SocketClientSink[String]("localhost", 5555, new SimpleStringSchema))