Druid 中更新订单数据的聚合

如何解决Druid 中更新订单数据的聚合

我有使用 Kafka 到 Druid 的流数据。这是一个电子商务非规范化订单事件数据，其中状态和几个字段在每个事件中都会更新。我只需要使用最新的条目进行基于时间戳的聚合查询。

例如：如果数据样本是：

{"orderId":"123","status":"Initiated","items":"item","qty":1,"paymentId":null,"shipmentId":null,timestamp:"2021 -03-05T01:02:33Z"} {"orderId":"abc",timestamp:"2021-03- 05T01:03:33Z"} {"orderId":"123","status":"Shipped",timestamp:"2021-03- 07T02:03:33Z"}

现在，如果我想查询所有停留在“已启动”状态超过 2 天的订单，那么对于上述数据，它应该只显示 orderId“abc”。但是如果我查询类似

Select orderId,qty,paymentId from order where status = Initiated and WHERE "timestamp" < TIMESTAMPADD(DAY,-2,CURRENT_TIMESTAMP)

此查询将同时返回订单“123”和“abc”，但 123 在 2 天后收到另一个事件，因此不应将之前的事件包含在结果中。

在 Apache druid 中，他们有什么好的和优化的方法来处理这种场景吗？

我想使用单独的查找表来存储 orderId 和最新状态并在 orderId 和 status 上执行此查找和以上聚合查询的连接的一种方法

编辑 1：此查询有效，但它连接到整个表，这可以为大数据集提供资源限制例外：

WITH maxOrderTime (orderId,"__time") AS
(
SELECT orderId,max("__time") FROM inline_data
   GROUP BY orderId
)
SELECT inline_data.orderId FROM inline_data 
JOIN maxOrderTime
ON inline_data.orderId = maxOrderTime.orderId
AND inline_data."__time" = maxOrderTime."__time"
WHERE inline_data.status='Initiated' and inline_data."__time" < TIMESTAMPADD(DAY,CURRENT_TIMESTAMP)

编辑 2：尝试过：

SELECT
  inline_data.orderID,MAX(LOOKUP(status,'status_as_number')) as last_status
FROM inline_data
WHERE
  inline_data."__time" < TIMESTAMPADD(DAY,CURRENT_TIMESTAMP)
GROUP BY inline_data.orderID
HAVING last_status = 1

但是给出了这个错误：

错误：未知异常

应用规则时出错 DruidQueryRule(AGGREGATE),args [rel#1853:LogicalAggregate.NONE.,rel#1863:DruidQueryRel.NONE.[](query={"queryType":"scan","dataSource":{"type":"table","name":"inline_data"},"intervals":{" type":"intervals","intervals":["-146136543-09-08T08:23:32.096Z/2021-03-14T09:57:05.000Z"]},"virtualColumns":[{"type":" expression","name":"v0","expression":"lookup("status",'status_as_number')","outputType":"STRING"}],"resultFormat":"compactedList","batchSize": 20480,"order":"none","filter":null,"columns":["orderID","v0"],"legacy":false,"context":{"sqlOuterLimit":100,"sqlQueryId" :"fbc167be-48fc-4863-b3a8-b8a7c45fb60f"},"降序":false,"粒度":{"type":"all"}},signature={orderID:LONG,v0:STRING})]

java.lang.RuntimeException

解决方法

我认为这可以更容易地完成。如果将状态替换为数字表示，则可以更轻松地使用它。

首先使用内联查找来替换状态。请参阅此页面如何定义查找：https://druid.apache.org/docs/0.20.1/querying/lookups.html

现在，例如，我们在名为 status_as_number 的查找中有这些值：

Initiated = 1
Shipped = 2

由于我们现在有一个数字表示，您可以简单地按查询进行分组并查看最大状态数。像这样的查询就足够了：

SELECT
  inline_data.orderId,MAX(LOOKUP(status,'status_as_number')) as last_status
FROM inline_data
WHERE
  inline_data."__time" < TIMESTAMPADD(DAY,-2,CURRENT_TIMESTAMP)
GROUP BY inline_data.orderId
HAVING last_status = 1

注意：此查询未经测试。 HAVING 部分确保您只能看到已启动的订单。

希望这能解决您的问题。