gpt4 book ai didi

java - 全局状态存储不创建更改日志主题如果全局存储的输入主题具有空键,解决方法是什么?

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

我读了很多关于全局状态存储的内容,它不会创建用于恢复的更改主题主题,而是使用源主题作为恢复。

我正在创建自定义 key 并将数据存储在全局状态存储中,但重新启动后它将消失,因为恢复时的全局存储将直接从源主题获取数据并绕过处理器。

我的输入主题有上述数据。

{
"id": "user-12345",
"user_client": [
"clientid-1",
"clientid-2"
]
}

我正在维护两个状态存储,如下所示:

  1. id ->record(记录表示上面的json)
  2. clientid-1:[“user-12345”](clientid -> 用户 ID)
  3. clientid-2:[“user-12345”](clientid -> 用户 ID)

因此,我看到的解决方法是创建一个自定义更改日志主题,并向该主题发送带有 key 的数据,该主题将充当全局状态存储的源主题。

但在我的场景中,我必须在状态存储中填写两条记录,最好的方法是什么。

示例场景:

Record1: {
"id": "user-1",
"user_client": [
"clientid-1",
"clientid-2"
]
}



Record2:{
"id": "user-2",
"user_client": [
"clientid-1",
"clientid-3"
]
}

全局状态存储应该具有:

id -> json Record'

clientid-1: ["user-1", "user-2"]
clientid-2: ["user-2"]
clientid-3: ["user-2"]

如何在全局状态存储中维护上述场景的恢复情况

最佳答案

一种方法是我们为 GlobalKTable 维护一个变更日志主题(具有retention.policy=compact),我们将其称为user_client_global_ktable_changelog,为了简单起见,假设我们将您的消息序列化为java类(你可以只使用 HashMap 或 JsonNode 或其他东西):

//initial message format
public class UserClients {
String id;
Set<String> userClient;
}
//message when key is client
public class ClientUsers {
String clientId;
Set<String> userIds;
}
//your initial topic
KStream<String, UserClients> userClientKStream = streamsBuilder.stream("un_keyed_topic");
  1. 将记录重新键入 user_id 很容易,只需重新键入 KStream 然后将其发送到输出主题
//re-map initial message to user_id:{inital_message_payload}
userClientKStream
.map((defaultNullKey, userClients) -> KeyValue.pair(userClients.getId(), userClients))
.to("user_client_global_ktable_changelog");//please provide appropriate serdes
  • 聚合特定客户端的 user_id,我们可以使用本地状态 (KTable) 来保存(当前 client_id 的当前 user_ids 列表):
  • userClientKStream
    //will cause data re-partition before running groupByKey (will create an internal -repartition topic)
    .flatMap((defaultNullKey, userClients)
    -> userClients.getUserClient().stream().map(clientId -> KeyValue.pair(clientId, userClients.getId())).collect(Collectors.toList()))
    //we have to maintain a current aggregated store for user_ids for a particular client_id
    .groupByKey()
    .aggregate(ClientUsers::new, (clientId, userId, clientUsers) -> {
    clientUsers.getUserIds().add(userId);
    return clientUsers;
    }, Materialized.as("client_with_aggregated_user_ids"))
    .toStream()
    .to("user_client_global_ktable_changelog");//please provide appropriate serdes

    例如,用于聚合本地状态中的 user_ids:

    //re-key message for client-based message
    clientid-1:user-1
    //your current aggregated for `clientid-1`
    "clientid-1"
    {
    "user_id": ["user-1"]
    }

    //re-key message for client-based message
    clientid-1:user-2
    //your current aggregated for `clientid-1`
    "clientid-1"
    {
    "user_id": ["user-1", "user-2"]
    }

    实际上,如果您进行一些更改,我们可以直接使用本地状态的更改日志主题作为 GlobalKTable 的更改日志,即主题 your_application-client_with_aggreated_user_ids-changelog,通过调整状态以保留负载用户 key 和客户端 key 消息。

    关于java - 全局状态存储不创建更改日志主题如果全局存储的输入主题具有空键,解决方法是什么?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/60613596/

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