0

我有一个Flink流媒体系统，我可以在其中获取点击流数据。

数据格式：

{"uid":"123", "event_type":"view","payload":{"p1":{"price":23}}}
{"uid":"123", "event_type":"view","payload":{"p2":{"price":25}}}
{"uid":"123", "event_type":"a2c","payload":{"p2"}}
{"uid":"123", "event_type":"a2c","payload":{"p1":{}}}

所以，在这里我们看到我们得到两种类型的事件a2c和view. 这两种类型的事件之间的区别在于view所有产品的事件价格都存在，并且a2c只有名称存在，价格不存在。

我想要实现的是在 10 分钟的窗口中从所有事件中创建一个合并的有效负载。还通过从其相应事件中a2c获取信息来丰富有效负载。priceview

每个 uid 的窗口完成后合并的有效负载：

{
    "uid":"123",
    "all":[
        {"event_type":"view", "payload":{"p1":{"price":23}}},
        {"event_type":"view","payload":{"p2":{"price":25}}},
        {"event_type":"a2c","payload":{"p2":{"price":25}}},
        {"event_type":"a2c","payload":{"p1":{"price":23}}}
    ],
   "total":4
}

我怎样才能做到这一点？基本上我必须维护view窗口中所有事件的状态，然后一旦我得到a2c事件，我必须从状态中获取价格。我不是要求任何可行的解决方案，只是如何维护view窗口中所有事件的状态。我也有一些自定义的 Reduce 操作。

events.keyBy("uid").window(..).reduce(new ReduceCustomFun(..)).uid("..").name("..");

在ReduceCustomFun：我将 2 个事件数据加入列表中。

4

1 回答 1

0

最简单的做法是使用ProcessWindowFunction在每 10 分钟窗口结束时执行所有窗口处理的 a。然后，您将拥有一个Iterable包含所有事件（对于给定的 uid）来创建合并报告的事件，并且您不必担心维护任何状态。

于 2020-03-29T15:06:05.030 回答