Apache Flink动态添加新流

Question

val env = StreamExecutionEnvironment.getExecutionEnvironment()
val text = env.socketTextStream(hostname, port, "\n")
val windowCounts = text.map...

env.execute("Socket Window WordCount")

在我的情况下，有可能，例如，一个新的设备被启动，因此必须处理另一个流。但是如何动态添加这个新流呢？

Answer 1

不可能在运行时向Flink程序添加新流。

解决这个问题的方法是拥有一个包含所有传入事件的流（例如，一个Kafaka主题，您将所有单独的流摄入其中）。事件应该有一个标识它们来自哪个流的键。然后可以使用此键来keyby流并应用每键处理逻辑。

如果您想从多个套接字读取，那么您可以编写自己的sourcefunction，它从某些输入（例如，从固定套接字）读取要为其打开套接字的端口。然后，在内部，您可以保持所有这些套接字的打开，并以循环方式读取它们。