0

我的用例 - 收集特定持续时间的事件，然后根据键对它们进行分组

Objective 处理后，用户可以根据key保存特定时长的数据

我打算怎么做 1）从 Kafka 接收事件

2）创建事件数据流

3）通过运行 SQL 查询将表与其关联并收集特定持续时间的数据

4）将新表与第二步输出相关联，并根据键对收集的数据进行分组

5）将数据保存在数据库中

我试过的解决方案-

我能够-

1）从卡夫卡接收事件，

2）设置数据流（比如说sensorDataStream）-

DataStream<SensorEvent> sensorDataStream 
         = source.flatMap(new FlatMapFunction<String, SensorEvent>() {
            @Override
            public void flatMap(String catalog, Collector<SensorEvent> out) {
            // create SensorEvent(id, sensor notification value, notification time) creation
             });

3）将表（比如说table1）与数据流相关联，并在运行SQL查询之后 -

SELECT id, sensorNotif, notifTime FROM SENSORTABLE WHERE notifTime > t1_Timestamp AND notifTime < t2_Timestamp.

这里 t1_Timestamp 和 t2_Timestamp 是预定义的纪元时间，会根据一些预定义的条件而改变

4）我可以通过在控制台上使用以下查询来打印此 sql 查询结果-

tableEnv.toAppendStream(table1, Row.class).print();

5）使用table1和以下类型的sql查询创建了一个新表（比如说table2）-

Table table2 = tableEnv.sqlQuery("SELECT id AS SensorID, COUNT(sensorNotif) AS SensorNotificationCount FROM table1 GROUP BY id);

6）通过使用收集和打印数据 -

tableEnv.toRetractStream(table2 , Row.class).print();

问题

1）我无法在控制台上看到第 6 步的输出。

我做了一些实验，发现如果我跳过 table1 设置步骤（这意味着一段时间内没有传感器数据俱乐部）并将我的 senserDataStream 与 table2 直接关联，那么我可以看到第 6 步的输出，但因为这是 RetractStream 所以我可以看到形式的数据，如果新事件即将到来，则此撤回流将使数据无效并打印新计算的数据。

我想要的建议

1）如何合并步骤 5 和步骤 6（表示表 1 和表 2）。我已经合并了这些表，但是由于控制台上看不到数据，所以我有疑问？难道我做错了什么？还是数据已合并但不可见？

2）我的计划是——

2.a) 在 2 遍中过滤数据，在第一遍中过滤特定时间间隔的数据，在第二遍中将数据分组

2.b）在数据库中保存 2.a 输出 这种方法是否有效（我有疑问，因为我正在使用数据流并且 table1 输出是附加流但 table2 输出是撤回流）？

4

0 回答 0