如何组织复杂的Apache应用程序？

Question

我们使用flink从一些IoT传感器生成事件。每个传感器都可以用来产生不同类型的事件(如温度、湿度等)。一对多的比率(传感器->启用事件)。

存储在关系数据库中的传感器和启用事件之间的映射

为了丰富传感器数据，我们将传感器数据和表API连接起来。只需添加带有启用事件列表的元数据。

因此，如果某些特定的sensor-123只启用了TEMP和PRESSURE事件，如何仅将传感器数据发送到这两个已定义的进程函数？

人们会想到以下几点：

val enriched: DataStream[EnrichedSensorData] = ...

val temp = enriched.filter(x => isTempEnabled(x)).process(....)
val humd = enriched.filter(x => isHumdEnabled(x)).process(....)
val press = enriched.filter(x => isPressEnabled(x)).process(....)

，

1. ，它有多有效？如何在flink最佳实践方面做得最好？据我所知，在我的例子中，我多次乘以数据流，尽管我随后过滤了结果

。

1. 在我的例子中，做数据丰富过程的最好方法是什么？将传感器数据流连接到表流(通过flink-cdc-连接器)+在充实处理功能中使用状态缓存映射sensorId -> ->

​

​

Answer 1

1. 使用来自丰富函数的侧输出来生成三个事件流。如果您有与复制数据有关的性能问题，您可以尝试将其流水线化(将温度、湿度和压力函数内联起来，并转发任何不适合处理的记录)。

1. 如果您有数百万个传感器，每个传感器都包含元数据，那么使用JDBC源，并对传感器数据执行一个(有状态)连接。您必须在相应的元数据记录之前处理获取传感器数据记录的情况，在这种情况下，您需要将其存储在状态中，然后在元数据记录到达时生成结果(和清除状态)。