温馨提示:本文翻译自stackoverflow.com，查看原文请点击：java - Kafka streams creating a simple materialized view
java apache-kafka-streams spring-kafka

java - Kafka流创建简单的物化视图

发布于 2020-03-27 11:51:09

我有一些非唯一的String字段和事件时间戳进入Kafka。我想创建这些事件的实例化视图，以便可以查询它们。例如：

  1. 显示所有事件
  2. 显示所有事件 field1 = some string
  3. 显示与多个字段匹配的所有事件
  4. 显示两个日期之间的事件

我看到的所有示例在流上都有聚合，联接或其他一些转换操作。我找不到在事件集上创建视图的单个简单示例。我不想执行任何操作，只希望能够查询流中的原始事件。

我使用的Spring Kafka是一个Spring Kafka理想的例子。

我能够将消息发送到Kafka并进行使用。但是，我无法创建实例化视图。

我有以下用于过滤事件的代码（不是我真正想要的，我想要所有事件，但我只是想看看是否可以获得实例化视图）：

@StreamListener
    public void process(@Input("input") KTable<String,MyMessage> myMessages) {
        keyValueStore = interactiveQueryService.getQueryableStore(ALL_MESSAGES,QueryableStoreTypes.keyValueStore());

        myMessages.filter((key,value) -> (value.getKey() != null));
                Materialized.<String,MyMessage,KeyValueStore<Bytes,byte[]>> as(ALL_MESSAGES)
                .withKeySerde(Serdes.String())
                .withValueSerde(new MyMessageSerde());

这引发了异常：

java.lang.ClassCastException: [B cannot be cast to MyMessage
at org.apache.kafka.streams.kstream.internals.KTableFilter.computeValue(KTableFilter.java:57)
    at org.apache.kafka.streams.kstream.internals.KTableFilter.access$300(KTableFilter.java:25)
    at org.apache.kafka.streams.kstream.internals.KTableFilter$KTableFilterProcessor.process(KTableFilter.java:79)
    at org.apache.kafka.streams.kstream.internals.KTableFilter$KTableFilterProcessor.process(KTableFilter.java:63)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
    at org.apache.kafka.streams.kstream.internals.ForwardingCacheFlushListener.apply(ForwardingCacheFlushListener.java:42)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putAndMaybeForward(CachingKeyValueStore.java:101)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.access$000(CachingKeyValueStore.java:38)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore$1.apply(CachingKeyValueStore.java:83)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:141)
    at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:99)
    at org.apache.kafka.streams.state.internals.ThreadCache.flush(ThreadCache.java:125)
    at org.apache.kafka.streams.state.internals.CachingKeyValueStore.flush(CachingKeyValueStore.java:123)
    at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.flush(InnerMeteredKeyValueStore.java:284)
    at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.flush(MeteredKeyValueBytesStore.java:149)
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:239)
    ... 21 more

我不明白为什么，因为我将商店的valueSerde设置为MyMessageSerde，它知道如何将MyMessage序列化/反序列化为字节数组。

更新资料

我将代码更改为以下内容：

myMessages.filter((key,value) -> (value.getKey() != null));

并将以下内容添加到我的application.yml中 

spring.cloud.stream.kafka.streams.bindings.input:
  consumer:
    materializedAs: all-messages
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: MyMessageDeserializer  `

现在，我得到以下堆栈跟踪：

Exception in thread "raven-a43f181b-ccb6-4d9b-a8fd-9fe96542c210-StreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [0_3] Failed to flush state store all-messages
at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:242)
at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:202)
at org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:420)
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:394)
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:382)
at org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:67)
at org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:362)
at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:352)
at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:401)
at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1042)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:845)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736)
Caused by: java.lang.ClassCastException: [B cannot be cast to MyMessage
at org.apache.kafka.streams.kstream.internals.KTableFilter.computeValue(KTableFilter.java:57)
at org.apache.kafka.streams.kstream.internals.KTableFilter.access$300(KTableFilter.java:25)
at org.apache.kafka.streams.kstream.internals.KTableFilter$KTableFilterProcessor.process(KTableFilter.java:79)
at org.apache.kafka.streams.kstream.internals.KTableFilter$KTableFilterProcessor.process(KTableFilter.java:63)
at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50)
at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90)
at org.apache.kafka.streams.kstream.internals.ForwardingCacheFlushListener.apply(ForwardingCacheFlushListener.java:42)
at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putAndMaybeForward(CachingKeyValueStore.java:101)
at org.apache.kafka.streams.state.internals.CachingKeyValueStore.access$000(CachingKeyValueStore.java:38)
at org.apache.kafka.streams.state.internals.CachingKeyValueStore$1.apply(CachingKeyValueStore.java:83)
at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:141)
at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:99)
at org.apache.kafka.streams.state.internals.ThreadCache.flush(ThreadCache.java:125)
at org.apache.kafka.streams.state.internals.CachingKeyValueStore.flush(CachingKeyValueStore.java:123)
at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.flush(InnerMeteredKeyValueStore.java:284)
at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.flush(MeteredKeyValueBytesStore.java:149)
at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:239)
... 12 more`

查看更多

查看更多

提问者
mbluke
被浏览
54
分享
Matthias J. Sax 2019-07-06 05:09

所需的查询类型不容易得到支持。请注意，没有二级索引，但仅支持基于常规键的查找和范围。

如果您预先知道所有查询，则可以将数据重新分组为KTables以查询属性为键的派生数据。请注意，键必须是唯一的，因此，如果查询属性包含非唯一数据，则需要使用某种Collection类型作为值：

KTable originalTable = builder.table(...)
KTable keyedByFieldATable = originalTable.groupBy(/*select field A*/).aggregate(/* the aggregation return a list or similar of entries for the key*/);

请注意，您每次重新键入原始表时都会重复存储需求。

或者，您可以对原始表进行全表扫描，并在使用返回的迭代器时评估过滤条件。

这是空间与CPU之间的权衡。也许Kafka Streams不是解决您问题的正确工具。

热门帖子

1
为什么我的 TG 始终收不到提醒通知？
2
Android 开发方向：传统 View 开发 or 拥抱 Jetpack Compose
3
浙江 ISP 会屏蔽 Wireguard UDP 端口
4
高级 Java 岗位一定需要有管理经验吗
5
兄弟们，老是倒在大厂二面的怎么办？
6
最近很多小伙伴弄港卡，分享一下个人开卡经历
7
vercel 免费版 3 个指标都超了，好像我的网站已经获得了基础流量。还好套了 cloudflare 不然流量也超了。
8
最近换了真我手机，发现这系统连应用的数据都无法备份
9
MacBook Pro 决赛圈
10
连下厨房也要放弃网页版了

相关问题

1
检查数据类型以构建对象的替代方法
2
是否可以从Firebase上的Google身份验证获取姓氏和名字？
3
接收TXT文件以读取和存储它们的构造方法
4
xml DSL中的 Camel 谓词示例
5
如何获得用户输入验证到我的Java计算器程序中？
6
Firebase从数据库中获取确切的孩子
7
自定义对话框的侦听器null
8
带有Logback的AWS Elastic Beanstalk应用程序日志记录
9
如何更改ArrayList中对象的显示名称
10
Flutter BLE字节到阵列的转换

热门github

1
A multi-platform library for OpenGL, OpenGL ES, Vulkan, window and input (翻译：适用于 OpenGL、OpenGL ES、Vulkan、窗口和输入的多平台库)
2
Dev tool that writes scalable apps from scratch while the developer oversees the implementation (翻译：可扩展开发工具的 PoC，该工具从头开始编写整个应用程序，同时开发人员监督实施)
3
shadcn/ui, but for Svelte. ✨ (翻译：shadcn-svelte是shadcn/ui的非官方社区主导的Svelte端口。)
4
The Python Risk Identification Tool for generative AI (PyRIT) is an open access automation framework to empower security professionals and machine learning engineers to proactively find risks in their generative AI systems. (翻译：用于生成式 AI 的 Python 风险识别工具 (PyRIT) 是一个开放式访问自动化框架，使安全专业人员和机器学习工程师能够主动发现其生成式 AI 系统中的风险。)
5
Performance-portable, length-agnostic SIMD with runtime dispatch (翻译：Highway 是一个提供可移植 SIMD/向量内在函数的 C++ 库。)
6
ZK Credo (翻译：ZK信条)
7
OpenCodeInterpreter: Integrating Code Generation with Execution and Refinement (翻译：OpenCodeInterpreter：将代码生成与执行和优化集成)
8
Joplin - the secure note taking and to-do app with synchronisation capabilities for Windows, macOS, Linux, Android and iOS. (翻译：Joplin - 一个开源的笔记和待办事项应用程序，具有Windows，macOS，Linux，Android和iOS的同步功能。)
9
Mamba is a new state space model architecture showing promising performance on information-dense data such as language modeling, where previous subquadratic models fall short of Transformers. It is based on the line of progress on structured state space models, with an efficient hardware-aware design and implementation in the spirit of FlashAttention. (翻译：Mamba 是一种新的状态空间模型架构，在信息密集型数据（例如语言建模）上显示出良好的性能，而之前的二次模型在 Transformers 方面存在不足。它基于结构化状态空间模型的进展，并本着FlashAttention的精神进行高效的硬件感知设计和实现。)
10
This repository contains System Design resources which are useful while preparing for interviews and learning Distributed Systems (翻译：该存储库包含系统设计资源，在准备面试和学习分布式系统时非常有用)
11
Curso para aprender el lenguaje de programación Python desde cero y para principiantes. 75 clases, 37 horas en vídeo, código, proyectos y grupo de chat. Fundamentos, frontend, backend, testing, IA... (翻译：从零开始学习 Python 编程语言的课程，适合初学者)
12
🎓 Path to a free self-taught education in Computer Science! (翻译：🎓计算机科学免费自学教程！)
13
1️⃣🐝🏎️ The One Billion Row Challenge -- A fun exploration of how quickly 1B rows from a text file can be aggregated with Java (翻译：十亿行挑战 —— 使用 Java 对文本文件中的 10 亿行数据进行聚合的有趣探索)
14
A collective list of free APIs (翻译：免费 API 的集合列表)
15
📚 Freely available programming books (翻译：📚 免费提供的编程书籍)