使用KStreams合并记录

文章/答案/技术大牛

发布

1回答

java、apache-kafka、apache-kafka-streams

我正在使用Java处理我的第一个Kafka Stream，基本上我是在尝试从一个主题中读取记录，然后将这些记录的值合并到一个单独的记录中，以便写入到单独的主题中，即将一系列记录批量处理成一个单独的记录理想情况下，此批处理将由记录大小或记录数量控制，但此时将很高兴获得一个简单的示例！我认为KStream.groupBy函数将是解决方案的基础，但无法提供一个可行的示例

浏览 14提问于2019-01-22得票数 0

1回答

KStream到KTable内连接，每次使用相同的数据处理时产生不同数量的记录

apache-kafka、apache-kafka-streams、spring-kafka、spring-kafka-test

使用KTable作为一个查找表。，KTable中有14条记录，并且假设KStreams中的每个键都有一个KTable记录。我尝试使用测试用例，使用嵌入式卡夫卡进行KStream和Ktable。其中KStreams记录10条，KTable记录3条。当我第一次运行测试用例时，没有join，也没有在Join之后看到任何数据。我尝试了使用KSQL，join运行得很好，我得到了8000条记录，然后我进入了KS

浏览 5提问于2020-03-08得票数 0

回答已采纳

1回答

考虑Kafka流和Spring流的设计问题

apache-kafka、apache-kafka-streams、spring-kafka、spring-cloud-stream

我需要维护外部系统记录(KTables)，并跟踪这些记录上的任何更改(KStreams)。我很感激任何

浏览 4提问于2021-03-09得票数 0

1回答

KStream-KStream在Kafka流中加入的非重复中间结果

apache-kafka、apache-kafka-streams、debezium

我有以下情况： ie.最终目标是能够识别最新的聚合，以便在查询

浏览 0提问于2019-05-10得票数 1

回答已采纳

1回答

kafka能捕获数据库视图中的更改吗？

database、apache-kafka、streaming

卡夫卡流是否有可能捕捉到数据库视图中的变化？我在数据库中有一个包含多个表的列的视图。卡夫卡也会在视野中检测到数据的变化吗？

浏览 0提问于2018-11-30得票数 1

回答已采纳

1回答

Kafka KStream to KStream join |重启性能

performance、caching、apache-kafka、apache-kafka-streams、kafka-join

我计划在很长一段时间内(~1周)以KStreams的身份加入两个主题。假设在此窗口中累积了数亿条记录，加入的消费者需要多长时间才能重启？我之所以这样问，是因为我无法找到有关窗口中有多少记录存储在使用者缓存中的信息。

浏览 7提问于2020-01-03得票数 0

回答已采纳

1回答

当我加入卡夫卡流时，我怎样才能访问原始记录？

java、apache-kafka、apache-kafka-streams

我有一个卡夫卡流应用程序，目前正在创建两个不同主题的KStreams。那部分运作得很好。我只想要一个<key, merge value1 + value2>的输出流，用于在这两个流上都有相同键的记录。我可以手动合并这些值，但是还不清楚如何访问lambda右侧的值。

浏览 0提问于2019-04-15得票数 1

回答已采纳

2回答

如果所有代理都不可用，如何关闭KStream

java、spring-boot、apache-kafka、apache-kafka-streams、spring-kafka

我在SpringBoot应用程序中使用Kstreams。如果卡夫卡已关闭或无法到达，则仅锁定以下内容：无法建立到节点-1的连接。可能找不到经纪人。如果出现这种情况，我希望关闭应用程序，或者至少记录异常。有没有办法做同样的事。

浏览 0提问于2019-01-16得票数 1

回答已采纳

1回答

Kafka KStreams中的“‘for”循环支持

apache-kafka、apache-kafka-streams

我需要知道如何在我的卡夫卡KStreams line...below是我的'for‘循环中使用'for’循环，这个循环需要包含在KStreams中。}catch (InterruptedException e){ }我的KStreams> mstream = textlines .mapValues(value -> value.replac

浏览 0提问于2017-12-18得票数 0

回答已采纳

1回答

卡夫卡:如何查询主题的前一状态

apache-kafka

作为关键，我使用书籍的类别(例如“科学”、“幻想”、“外国”)。晚上9点。

浏览 0提问于2019-07-22得票数 0

回答已采纳

1回答

如何扩大卡夫卡流？

apache-kafka-streams

我有一个使用Apache流的应用程序。对于卡夫卡消费者，同样的consumer.group.id保证在不同的情况下(即不同的机器)不同的应用被视为相同的消费者群体。

浏览 0提问于2018-03-12得票数 1

回答已采纳

1回答

我可以使用KSQL来生成处理超时吗？

apache-flink、apache-kafka-streams、ksqldb

我正在尝试使用KSQL在一个时限内进行任何处理，并在那个时间限制中得到结果。请参阅“处理时间定时器”下面的，以获得使用Apache演示的相同想法。从概念上讲，我正在考虑创建第一个流的KTable，以保存事务的最新状态，并使用KSQL创建一个输出流，方法是使用(create_time +超时值)< current_time查询KTable中的键。并将超时作为“更新”添加到第一个流中，这样我就可以从KTable中筛选出这些超时) 我还没有在KSQL文档中找到这样的方法，即使有一个内置的current_time，我也不确定

浏览 1提问于2018-11-14得票数 1

回答已采纳

2回答

KStream将记录发送到多个流(不是分支)

apache-kafka、apache-kafka-streams、spring-kafka

有没有一种方法可以进行类似分支的操作，但将记录放在谓词计算为true的每个输出流中？分支将记录放到第一个匹配项(文档:在第一个匹配项上，一条记录被放置到且只有一个输出流)。

浏览 29提问于2019-01-06得票数 2

回答已采纳

1回答

Kafka中的动态流拓扑

java、apache-kafka、apache-kafka-streams

使用Kafka流DSL 创建Kafka流时例如:当我们开始的时候，我们定义了一个拓扑来读取一个主题(来源)和一个目标主题(Sink)。是否有可能重新加载一个现有的Kafka流使用一个新的定义卡夫卡流拓扑？我们使用JDK 1.8和Kafka DSL 2.2.0 谢谢你，阿尤斯曼

浏览 0提问于2019-06-18得票数 1

回答已采纳

1回答

有没有一种方法可以根据大小来冲洗卡夫卡流WindowStore？

apache-kafka-streams

我正在使用DSL编写Kafka应用程序，它将从kafka主题中读取元组。在拓扑结构中，我想对元组进行批处理。我编写的拓扑使用TimeWindowedKStream对元组进行分组。然后调用聚合并传递一个窗口存储。org.apache.kafka.streams.errors.StreamsException:任务ibv2-capt-consumer-group-3-record-store-changelog中止发送，因为以前记录捕获的错误这是我的拓扑这是我一直在使用的Scala代码。注意，我这里

浏览 0提问于2019-01-07得票数 0

2回答

如何将记录分成不同的流，从一个主题到不同的流？

apache-kafka、apache-kafka-streams、apache-kafka-connect

我有一个源CSV文件，包含不同大小的记录，将每条记录推入一个源主题。我想把记录分成不同的KStreams/KTables与那个源主题。我有一个用于一个表加载的管道，在该管道中，我以分隔格式将记录从源主题推入stream1，然后将记录按AVRO格式推到另一个流中，然后将该流推入JDBC接收器连接器，该连接器将记录推入MySQL数据库。但我希望将不同表的记录推入一个源主题，然后根据一个值将记录拆分为不同的流。这个是可能的吗？我试

浏览 1提问于2020-04-13得票数 0

回答已采纳

1回答

KStreams :如何获取记录的(原始)主题？

java、apache-kafka、apache-kafka-streams

我有以下几点 //Config setupKStream<String, String> source = builder.stream(topicList); .map((k,v) -> { //How can i

浏览 21提问于2019-06-01得票数 1

回答已采纳

1回答

在Kstreams中创建和访问更新表

apache-kafka、apache-kafka-streams

从文档中我观察到，对于流数据"A“，我们可以使用Kstreams，而对于更新表"B”，我们可以使用Ktable。但是，只有kstreams和ktable之间允许的操作才是联接，并且它不更新ktable(相反，只使用它作为查找)。此外，它还创建了另一个新的kstream(我不想要)。

浏览 5提问于2017-11-21得票数 0

回答已采纳

1回答

如何用春云功能来消费卡夫卡事件，但有时才会产生。

kotlin、spring-kafka、spring-cloud-stream、spring-cloud-function

ProducerEvent { }现在，如果主题中有消息，则调用consume()-function，如果消息是针对我的，我将做一些事情，并使用我的结果调用

浏览 2提问于2020-07-29得票数 2

回答已采纳

4回答

KStreams +火花流+机器学习

apache-spark、machine-learning、apache-kafka、spark-streaming、apache-kafka-streams

我最初的想法是获取数据，使用但我能穿过KStreams。现在我糊涂了！我如何才能与KStreams +火花流+机器学习结合起来？我的想法是持续训练测试数据，而不是批量训练。

浏览 2提问于2016-12-14得票数 9

点击加载更多