我们希望使用Kafka Streams来创建和维护聚合，然后创建具有原始事务字段和聚合的传入事务的增强版本。此增强的记录由下游规则系统处理。TimeWindows twoDayHopping TimeWindows.of(TimeUnit.DAYS.toMillis(2)) })
.through(Serdes.String(),metricsSerde,da
浏览 3提问于2017-06-19得票数 0
回答已采纳
我使用Kafka Streams进行TimeWindow的一些聚合。我只对每个窗口的最终结果感兴趣，所以我使用了.suppress()特性，该特性为其状态创建了一个changelog主题。此changelog主题的保留策略配置被定义为"compact“，据我所知，它将至少保留过去每个键的最后一个事件。 我的应用程序中的问题是密钥经常改变。这意味着主题将无限增长(每个窗口都将带来永远不会删除的新关键字)。 因为聚合是按窗口进行
浏览 33提问于2019-02-12得票数 0
回答已采纳
我有许多IOT设备，它们通过消息向kafka主题报告事件，并且我已经定义了一个聚合器来从这些事件中更新设备状态。我想要做的是能够将输入流加入到聚合器在聚合更新状态之前输出的KTable --也就是说，我想，比方说，将一个事件与当前状态进行比较，如果它们匹配某个谓词，则执行一些处理，然后更新状态。我尝试将输入流加入到由KTable生成的StreamsBuilder#
浏览 0提问于2019-06-26得票数 0
回答已采纳
我们希望使用Flink来维护窗口聚合，作为事务监视应用程序的一部分。它们将使用滑动窗口定义。例如：“过去5天的现金交易总额”。下面是我需要Flink应用程序做的工作: 1.通过读取历史聚合和构建windows 2来准备事务处理。对于每个新事务: a.用新的事务数据b更新窗口聚合。找到与传入时间戳匹配的窗口，并将聚合值添加到事务c。通过RabbitMQ或Kafka接收器将增强的事务(原始字段+聚合)发
浏览 5提问于2017-06-23得票数 0
我试图:从Kafka读取事件、小时聚合(sum、count等)。并且在新事件出现时立即将结果重新插入Cassandra，换句话说，在每个新事件上创建新记录或重新计算已经存在的记录，并立即将结果汇到Cassandra。但是任务在窗口间隔结束后(每1小时)将结果发送给Cassandra。我知道，它的工作如docs中所述：
与连续表上的其他聚合不同，窗口聚合不会发出中间结果，而是只发出最终结果，即窗口末
浏览 11提问于2022-09-26得票数 0
回答已采纳
在Flink中，我有一个带有键控事件流的Job (例如:每个键平均每天有10个事件)。它们被处理为基于事件时间的滑动窗口(例如:90天窗口大小和1天窗口滑动)。事件是从Kafka消费的，Kafka会持久化所有的事件历史(例如:最近3年)。 有时我想重新启动Flink:用于维护、bug处理等，或者使用已经包含事件历史的Kafka启动一个新的Flink实例。在这种情况
浏览 9提问于2019-06-05得票数 0
回答已采纳
1回答
我想聚合一个具有翻滚窗口的Kafka主题，并且只获得每个窗口的最后一个聚合值。 使用下面粘贴的代码，每次有新事件到达时，聚合值都会更新。因此，我得到的聚合值与传入事件的数量一样多。我想有一个主题，只有每个窗口的最后聚合值。有可能(使用ksqldb)实现这一点吗？
浏览 11提问于2021-08-13得票数 1
有一个Kafka流应用程序，它通过流连接执行窗口(使用原始事件时间，而不是挂钟时间)，例如1天。如果打开这个拓扑，并从一开始就重新处理数据(就像在lambda风格的体系结构中那样)，那么这个窗口会将旧数据保存在那里吗？使用Kafka Streams 2.5.0进行更新
浏览 6提问于2022-01-09得票数 0
我目前正在使用Kafka流式DSL抑制功能的日期窗口。我们可能会遇到这样一种情况，一些事件可能会很晚才发生，超过了宽限期。Apache flink确实提供了对这种非常晚的事件的控制，并想知道这种功能是否在流中可用。2)考虑到数百万事件流经系统，如何使用DSL- suppress将间歇
浏览 2提问于2019-05-15得票数 0
我开始探索使用变更数据捕获将遗留和商业应用程序(我不能修改)中的数据库变更转换为其他系统可以使用的事件。简化我的实际案例，假设将涉及两个表，一个是包含订单头详细信息的order，另一个是包含所请求的每个产品的详细信息的order_line。我目前的理解是，这两个表中的事件将发布到两个不同的kafka主题中，我应该使用kafka-stream或ksql来聚合它们。我已经看到有不同的选项来定义窗口，该<
浏览 2提问于2020-02-27得票数 1
我一直在尝试使用Streams DSL在Kafka中实现滑动窗口，但我无法做到。有没有人能帮我举个代码例子。我想要使用滑动窗口方法聚合特定时间段的值。
我搜索了融合的博客，但没有代码
浏览 3提问于2019-07-16得票数 1
我目前正在实现一个从Kafka主题流式传输数据的应用程序。
使用应用程序只运行一次批处理是常见的场景吗?例如，一天结束时，收集主题中的所有数据，执行一些聚合和转换等操作？这意味着在使用spark-submit启动应用程序后，所有这些内容将在一个批处理中执行，然后应用程序将被关闭。或者，spark stream构建是为了以连续的批处理运行无休止的、永久的流数据吗？
浏览 1提问于2018-11-29得票数 3
我刚刚开始使用Spring Cloud Stream Kafka。然后，计算出的聚合将通过请求-响应层在UI中提供。我想知道Spring cloud stream和/或Kafka中是否支持聚合</em
浏览 5提问于2017-01-06得票数 0
1回答
类似于Kafka的日志压缩，有很多用例要求它只保留对给定键的最后更新，并使用结果(例如，用于连接数据)。----------------------------A | 15 | foobeedoo
我可能会加入到另一条溪流中最好不要浪费支持的聚合步骤。我想我需要一个具有反向顺序的dropDuplicates()函数。请注意，这个问题是关于结构化流的，
浏览 0提问于2018-06-12得票数 5
回答已采纳
越来越多的文章将kafka作为事件存储，并将其用于使用cqrs和事件源构建的应用程序中。如何在kafka (作为事件存储库)查询某个聚合的事件，以完成写入端的操作？
浏览 1提问于2016-10-15得票数 4
我们正在构建一个flink用例，其中我们正在使用kafka主题，并根据平均、最大、最小阈值执行聚合和生成警报。当滚动事件时间Windows中有0个事件时，我们还需要为每个键发出通知。Queryable :在每个进程窗口函数中保存密钥。从外部应用程序查询状态，并在20分钟时间间隔过期后丢失密钥时发出警报。我们看到Queryable状态特性在未来将被废弃。使用处理时间Windows :如果下游应用程序将实时发送事件，则使用处理时间而不是事件时间将是一种选择
浏览 6提问于2022-05-10得票数 0
在进行窗口聚合时，我面临一个问题。我希望执行每个键的值之和，结果只有在窗口完成时才发送到输出主题。问题是“输入”主题中的每个事件都会产生一个事件到“输出”主题。只有在窗口完成后，我才希望将事件发布到输出主题。例如，如果窗口为1分钟，则每分钟每键发送一个事件。:任务1_中的异常未能刷新状态存储KSTREAM-还原状态存储-0000000003，原因是: java.lang.ClassCastExcep
浏览 3提问于2019-12-10得票数 1
当前设置:我有一个来自kafka的数据流。每条消息都有一个事件时间。我正在使用这些事件时间来进行窗口聚合，并使用水印规则来丢弃状态。输出模式为追加模式。目的:我需要让窗口聚合在到期时按顺序排列，这样我就可以按事件窗口的顺序处理这些事件。我希望窗口状态会因为我的滑动窗口而依次过期。2020-06-11 08:
浏览 1提问于2020-05-06得票数 2
我想通过我的聚合事件处理程序之一将事件发布到axon Kafka主题，因为我正在使用kafka作为我的事件总线。这样做的正确方法是什么？应该直接将消息推送到主题，还是可以使用AggregateLifecycle#apply(event)？我有多个事件从我的聚合中发布，通过其中一个事件处理程序，我想发布另一个事件。
浏览 13提问于2020-07-24得票数 0
2回答
我有一个用例，在这个用例中，应用程序的输入是成批的XML文件。例如，每晚的一批银行交易。我试图看看是否可以使用事件源来创建事件日志。使用分布式日志机制(如Kafka)的事件来源是否是基于批/文件的输入的有效方法？
将事件日志写入<em
浏览 3提问于2017-07-07得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
实时音视频活动推荐
腾讯云开发者
