gpt4 book ai didi

apache-kafka - 是否可以从 kafka 消息中获取消息 key 的最新值

转载 作者:行者123 更新时间:2023-12-04 00:01:16 26 4
gpt4 key购买 nike

假设我对同一个消息键有不同的值。

例如:

{
userid: 1,
email: user123@xyz.com }

{
userid: 1,
email: user456@xyz.com }

{
userid: 1,
email: user789@xyz.com }

在上面的这种情况下,我只想要用户更新的最新值,即“user789@xyz.com”。

我的 kafka 流应该只给我第三个值,而不是前两个值。

最佳答案

由于您没有指定特定的客户端,我将向您展示如何使用 ksqlDB 和新添加的函数 LATEST_BY_OFFSET 完成此操作。

首先,我用源数据填充主题:

kafkacat -b broker:29092 -P -t test_topic -K: <<EOF
1:{ "userid": 1, "email": "user123@xyz.com" }
1:{ "userid": 1, "email": "user456@xyz.com" }
1:{ "userid": 1, "email": "user789@xyz.com" }
EOF

然后在 ksqlDB 中,首先将其建模为事件流:

ksql> CREATE STREAM USER_UPDATES (USERID INT, EMAIL VARCHAR) WITH (KAFKA_TOPIC='test_topic', VALUE_FORMAT='JSON');

Message
----------------
Stream created
----------------

ksql> SET 'auto.offset.reset' = 'earliest'; [35/60]
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
ksql> SELECT ROWKEY, USERID, EMAIL FROM USER_UPDATES EMIT CHANGES LIMIT 3;
+---------+---------+-----------------+
|ROWKEY |USERID |EMAIL |
+---------+---------+-----------------+
|1 |1 |user123@xyz.com |
|1 |1 |user456@xyz.com |
|1 |1 |user789@xyz.com |

现在我们可以告诉 ksqlDB 获取这个事件流并直接给我们最新的值(基于偏移量),或者直接:

ksql> SELECT USERID, LATEST_BY_OFFSET(EMAIL) FROM USER_UPDATES GROUP BY USERID EMIT CHANGES;
+--------------------+--------------------+
|USERID |KSQL_COL_1 |
+--------------------+--------------------+
|1 |user789@xyz.com |

Press CTRL-C to interrupt

或者更有用的是,作为 ksqlDB 中的物化状态:

CREATE TABLE USER_LATEST_STATE AS 
SELECT USERID, LATEST_BY_OFFSET(EMAIL) AS EMAIL
FROM USER_UPDATES
GROUP BY USERID
EMIT CHANGES;

该表仍然由对 Kafka 主题的更改驱动,但可以直接查询当前状态,无论是现在(“拉查询”):

ksql> SELECT EMAIL FROM USER_LATEST_STATE WHERE ROWKEY=1;
+--------------------+
|EMAIL |
+--------------------+
|user789@xyz.com |
Query terminated
ksql>

或者作为状态演变的变化流(“推送查询”):
ksql> SELECT EMAIL FROM USER_LATEST_STATE WHERE ROWKEY=1 EMIT CHANGES;
+--------------------+
|EMAIL |
+--------------------+
|user789@xyz.com |

[ query continues indefinitely ]

asciicast

关于apache-kafka - 是否可以从 kafka 消息中获取消息 key 的最新值,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/60964935/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com