我想要聚合来自kafka的记录流。我知道在JAVA中有kafka流的实现，但我不熟悉JAVA，我想知道在python中有没有执行这样的任务的选择，与kafka流相比，它有多好？
浏览 22提问于2020-03-19得票数 0
不同之处在于，我使用使用kafka connect的SQL来读取topic中的记录。我以固定的间隔获取记录，但不知何故，我没有获得输出中的最后几条记录。例如，Kafka topic的最后一条记录的时间戳是2020-11-26T13:11:36.605Z，聚合值的最后一条时间戳是2020-11-26T12:59:59.999。我不明白为什么我没有得到topic中最后一条记录的聚合。请帮帮忙。这是我的代
浏览 0提问于2020-11-27得票数 0
1回答
我使用Kafka来处理消息。消息可以分成几个部分(它是一个复合消息)。例如，在流中，我可以有一个被分成三个部分的复合消息。换句话说，它将是Kafka流中的三个记录，但这是一个大消息。我想使用Kafka表在一个Kafka记录中合并复合消息的各个部分。合并后，将在数据库(Postgres)中插入一封邮件。每个部件都有部件数和总部件数。例如，如果我在流中有一条消息的三个
浏览 5提问于2018-08-01得票数 0
我有一些历史数据，每条记录都有自己的时间戳。我想阅读它们并将它们输入到kafka主题中，并使用kafka流以时间窗口的方式对它们进行处理。现在的问题是，当我创建kafka流时间窗口聚合处理器时，我如何告诉kafka使用记录中的时间戳字段来创建时间窗口，而不是实际的实时时间？
浏览 9提问于2018-02-19得票数 3
回答已采纳
1回答
当您不使用带窗口设置的kafka流时，聚合是如何工作的？在使用物化键值存储时，它似乎有某种窗口。我似乎无法找到它何时会创建一个新的聚合的阈值。这将帮助我决定是否需要为我的用例添加一个自定义窗口。
浏览 1提问于2021-03-11得票数 2
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我们有一个Kafka流聚合拓扑。我们需要控制changeLog主题的大小，以降低Kafka存储成本。因此，我们在拓扑中使用transformer (DSL API)来调度一个标点符号，该标点符号使用keyValueStore.delete()从stateStore中删除旧记录。我能够验证在delete之后，在标点符号的进一步预定触发器上，被删除的键不在状态存储中。但是它是否也从changeLog主题中删除了该记录呢？
浏览 3提问于2020-08-29得票数 2
我们使用带有时间窗口的kafka-stream聚合来计算事件的结束和。我们已经实现了我们的需求，但是我们有一个中间聚合结果的问题。在本例中，输入是带有记录KStream<String,Integer>的<K,V>: <A, 1>, <D, 5>, <A, 20>, <A, 300>。本例中的重点是使用key == A的记录。聚合计算输入的记录值之
浏览 0提问于2018-07-12得票数 1
根据KSQLDB文档，可以使用会话窗口根据时间戳对记录进行排序和聚合。我有一个想要按顺序将记录插入到MySQL中的用例。我在我的记录中有一个时间戳字段，我将其用作ROWTIME，然后在其上尝试会话窗口，并将其插入到将推送到主题中然后推送到RDS中的输出流中。但在输出流中，我无法根据时间戳对消息进行重新排序。示例-有两条记录-记录1在上午11:00AM，记录2在上午11:01，并且这
浏览 20提问于2020-05-07得票数 0
回答已采纳
1回答
据我所知：
所以，如果我计划只使用Python，那么Faust应该是更好的选择，如果我想拥有更广泛的兼容性()，那么应该使用Kafka +Kafka吗？
浏览 0提问于2021-01-31得票数 5
回答已采纳
我使用kafka流组件在30分钟的滑动窗口上构建聚合(和)，宽限期为2分钟。我正在按10,000个时间序列(组)进行处理。聚合使用禁用日志记录的持久状态存储。在这些情况下，聚合值在许多情况下正好反映了输入流中的一个记录值。最初，我使用的是启用默认记录缓存(cache.max.bytes.buffering为10 MB)的聚合和禁用的withNoBound()选项，以便在需要时继续分配更多内存。str
浏览 3提问于2021-04-28得票数 1
我试图删除值为null的记录，在下游的changelog中，我知道在statestore中，它们仅仅是空(墓碑)就被删除了，但是当您在KTable或Stream上进行聚合时，它们跳过null而不删除它。我需要想办法在聚合中设置一个删除标志，让Kafka知道记录可以被删除。property to true in application.properties
// if ("tru
浏览 3提问于2019-11-22得票数 1
回答已采纳
_iterator)文件"\Python\Python38-32\lib\site-packages\kafka\consumer\group.py"，第1115行，在"\Python\Python38=False)文件第654行中，在投票记录中=self._poll_once(#en0#，max_records，update_offsets=update_offsets)文件recor
浏览 4提问于2020-05-15得票数 0
我是原型，计算聚合在一个星火结构流(Spark3.0)作业和发布的更新卡夫卡。我需要计算的最大日期和最大百分比的所有时间(没有窗口)的每一组。代码看起来很好，除非在源流中有Kafka墓碑记录(删除)。流接收一个具有有效键和空值的Kafka记录，但是最大聚合继续在计算中包含该记录。什么是最好的选择，让这个重新计算，而没有删除记录时，删除是从卡夫卡消费？":"2021-01-01 02:00:00&q
浏览 6提问于2021-01-19得票数 0
/kafka-topics.sh --zookeeper localhost:2181 --listorder-join-application3-address-store-name-changelog
浏览 4提问于2021-04-09得票数 0
1回答
我是Flink和Kafka的新手，有一些用Scala编写的数据聚合作业，这些作业在Apache Flink中运行，这些作业消耗Kafka中的数据，执行聚合并将结果返回给Kafka。我需要的作业，以消耗任何新的Kafka主题的数据，而作业正在运行，这是匹配的模式。我通过为我的消费者设置以下属性来实现这一点 val properties = new Properties()
properties.setProperty(“bootstra
浏览 10提问于2021-05-06得票数 1
我将页面访问和搜索查询(某种点击流分析)记录到kafka。我打算定期批处理它们，比如每小时或每天，然后将聚合结果写入elasticsearch。如果不是写入kafka，而是写入文件系统，则可以很容易地配置记录器写入文件名中有时间戳的文件，自动按照小时/天间隔划分日志。但对于kafka，我搜索了足够多的时间，还没有找到任何每小时/每天创建一个新主题的例子。如果不是每小时/每天创建一个新主题，我只能想象通过logstash/flume将日志从k
浏览 8提问于2017-07-20得票数 1
回答已采纳
confluent kafka是否为python语言中的流、分组和聚合提供了一个api？
浏览 2提问于2019-04-08得票数 1
回答已采纳
我刚刚开始使用Spring Cloud Stream Kafka。然后，计算出的聚合将通过请求-响应层在UI中提供。我想知道Spring cloud stream和/或Kafka中是否支持聚合</em
浏览 5提问于2017-01-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
