1

我正在尝试找到如何在 Kafka Streams 中对第n个事件执行操作的最佳方法。

我的情况：我有一个带有一些Events的输入流。我必须通过eventType == login过滤它们，并且在同一accountId的每个n次登录（比如说，第五次）上，将此Event发送到输出流。

经过一些调查和不同的尝试，我得到了下面的代码版本（我使用的是 Kotlin）。

data class Event(
    val payload: Any = {},
    val accountId: String,
    val eventType: String = ""
)

// intermediate class to keep the key and value of the original event
data class LoginEvent(
    val eventKey: String,
    val eventValue: Event
)

fun process() {
        val userLoginsStoreBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("logins"),
            Serdes.String(),
            Serdes.Integer()
        )
        val streamsBuilder = StreamsBuilder().addStateStore(userCheckInsStoreBuilder)
        val inputStream = streamsBuilder.stream<String, String>(inputTopic)

        inputStream.map { key, event ->
            KeyValue(key, json.readValue<Event>(event))
        }.filter { _, event -> event.eventType == "login" }
             .map { key, event -> KeyValue(event.accountId, LoginEvent(key, event)) }
             .transform(
                    UserLoginsTransformer("logins", 5),
                    "logins"
                )
             .filter { _, value -> value }
             .map { key, _ -> KeyValue(key.eventKey, json.writeValueAsString(key.eventValue)) }
             .to("fifth_login", Produced.with(Serdes.String(), Serdes.String()))

        ...
    }

class UserLoginsTransformer(private val storeName: String, private val loginsThreshold: Int = 5) :
    TransformerSupplier<String, CheckInEvent, KeyValue< LoginEvent, Boolean>> {

    override fun get(): Transformer<String, LoginEvent, KeyValue< LoginEvent, Boolean>> {
        return object : Transformer<String, LoginEvent, KeyValue< LoginEvent, Boolean>> {
            private lateinit var store: KeyValueStore<String, Int>

            @Suppress("UNCHECKED_CAST")
            override fun init(context: ProcessorContext) {
                store = context.getStateStore(storeName) as KeyValueStore<String, Int>
            }

            override fun transform(key: String, value: LoginEvent): KeyValue< LoginEvent, Boolean> {
                val counter = (store.get(key) ?: 0) + 1
                return if (counter == loginsThreshold) {
                    store.delete(key)
                    KeyValue(value, true)
                } else {
                    store.put(key, counter)
                    KeyValue(value, false)
                }
            }

            override fun close() {
            }
        }
    }
}

我最大的担心是transform在我的情况下函数不是线程安全的。我已经检查了在我的案例中使用的 KV 存储的实现，这是 RocksDB 存储（非事务性），因此值可能会在读取和比较之间更新，并且错误的事件将被发送到输出。

我的其他想法：

  1. 使用物化视图作为没有变压器的商店，但我坚持实施。
  2. 创建一个将使用 TransactionalRocksDB 的自定义持久 KV 存储（不确定是否值得）。
  3. 创建一个自定义的持久 KV 存储，该存储将在内部使用 ConcurrentHashMap（如果我们预期的用户很多，它可能会导致高内存消耗）。

还有一点需要注意：我正在使用 Spring Cloud Stream，所以也许这个框架有一个适合我的案例的内置解决方案，但我没有找到它。

我将不胜感激任何建议。提前致谢。

4

1 回答 1

3

我最担心的是转换函数在我的情况下不是线程安全的。我已经检查了在我的案例中使用的 KV 存储的实现，这是 RocksDB 存储（非事务性），因此值可能会在读取和比较之间更新，并且错误的事件将被发送到输出。

没有理由担心。如果您使用多个线程运行，每个线程将拥有自己的 RocksDB 存储整体数据的一个分片（请注意，整体状态是基于输入主题分区进行分片的，并且单个分片永远不会被不同的线程处理）。因此，您的代码将正常工作。您唯一需要确保的是，数据是按 分区的accountId，这样单个帐户的登录事件就会进入同一个分片。

如果您输入的数据在写入输入主题时已经被分区accountId，则您无需执行任何操作。如果没有，并且您可以控制上游应用程序，那么在上游的应用程序生产者中使用自定义分区器来获得您需要的分区可能是最简单的。如果您无法更改上游应用程序，则需要在设置accountId为新键后重新分区数据，即through()在调用之前执行 a transform()。

于 2019-12-08T00:59:24.080 回答