说明:我只想处理密钥的最新唯一事件。我有KafkaStreams kstreams。假设我在kafkaStreams中获得了以下事件： {id= "DELHI", event1},{id= "DELHI", event3},{id= "MUMBAI", event5} 现在，我想对它们进行分组(比如在10分钟内)，这样我就只有
浏览 0提问于2020-10-23得票数 0
1回答
我只是回顾了，以了解Google是如何处理水印的，它只是提到了非常模糊的内容：
我发现了一些东西，表明如果你的来源是谷歌PubSub，它已经有一个水印将被提取，但如果源是其他东西呢？例如，一个卡夫卡主题(我认为它本质上没有水印，所以我看不出像这样的东西会如何应用)。它是在最后几分钟确定最大滞后，如果是的话，有多少(肯定不会永远如此，因
浏览 6提问于2022-05-25得票数 0
或者它有更多的功能。然而，在这方面，我找不到一个好的例子。我在一个很好的解释工作逻辑的来源中也看不到它。你能解释一个很好的ktable和kstream的例子吗?我能做些什么？
浏览 0提问于2020-03-01得票数 0
1回答
我使用FluentD (最后一个稳定版本)向Kafka发送消息。但是FluentD使用的是旧的KafkaProducer，因此记录时间戳总是设置为-1。因此，当信息到达卡夫卡时，我必须使用WallclockTimestampExtractor将记录的时间戳设置为时间点。
“时间<
浏览 2提问于2017-10-12得票数 6
回答已采纳
我正在使用火花结构化流发送记录到一个卡夫卡主题。kafka主题是用config - message.timestamp.type=CreateTime创建的。这样做是为了使目标卡夫卡主题记录具有与原始记录相同的时间戳。我的卡夫卡流媒体代码：
kafkaRecords.selectExpr("CAST(key AS STRING)", "CAST(value AS BINARY)","C
浏览 0提问于2018-11-09得票数 0
我正在使用星火流和数据被发送给卡夫卡。我要给卡夫卡发地图。假设我有一个20的Map (在流批持续时间中它可能增长到1000 )元素，如下所示： KafkaProducer.send(record);我的Kafka主题是有10个分区。如何在</
浏览 0提问于2018-10-02得票数 1
WindowStore和ReadOnlyWindowStore方法fetch(K key, Instant from, Instant to)的Kafka流2.2.0文档声明：
对于每个键，迭代器保证窗口的顺序，从最老的/最早的可用窗口开始到最新/最新的窗口。其他的提取方法都没有声明这一点(除了不推荐的fetch(K key, long fr
浏览 0提问于2019-05-10得票数 0
回答已采纳
1回答
我想知道是否有可能提供输入来激发来自StreamSets的流。我注意到在StreamSets连接器目的地中不支持火花流。
我将探讨是否有其他方法将它们连接到一个示例POC。
浏览 3提问于2016-07-06得票数 3
回答已采纳
1回答
我正在使用SparkSession.readStream进行结构化流处理，并将其写入蜂窝表，但它似乎不允许我使用基于时间的微批，也就是说，我需要一批5秒的时间。所有的消息都应该形成一个5秒的批处理，并且批处理数据应该被写入到hive表中。
现在，当他们被发布到Kafka主题时，它会读取这些信息，每条信息都是表的一条记录。foreachBatch(hive_writ
浏览 12提问于2022-04-25得票数 0
回答已采纳
有一个Kafka流应用程序，它通过流连接执行窗口(使用原始事件时间，而不是挂钟时间)，例如1天。如果打开这个拓扑，并从一开始就重新处理数据(就像在lambda风格的体系结构中那样)，那么这个窗口会将旧数据保存在那里吗？例如:如果今天是2022-01-09，而我正在接收2021-03-01的数据，这个旧数据会进入表吗，还是会从一开始就被拒绝？
在这种情况下-可以采取什么策略来重新处理这些数据？使用Kafka Strea
浏览 6提问于2022-01-09得票数 0
2回答
我正经历着卡夫卡流的奇怪的任务行为。我有三个节点的卡夫卡流集群。我的流非常简单，有一个源主题(24个分区，所有kafka代理都运行在kafka流节点以外的其他机器上)，我们的流图只接收消息，按键对它们进行分组，执行一些筛选，并存储所有用于接收主题的内容。然而，每当我做滚动更新我的kafka流(通过关闭总是只有一
浏览 3提问于2021-10-09得票数 1
1回答
我使用的是合流3.3.0。我使用jdbc-source-connector将消息插入到我的Oracle表中的Kafka中。这个很好用。
我想检查一下是否可以“重新插入”。我的意思是，如果我有一个学生表，有3列id(数字)、name(varchar2)和last_modified(时间戳)。每当我插入新行时，它都会被推送到Kafka (使用timestamp+auto增量字段)。但是当我更新行时，应该更新Kafka<e
浏览 1提问于2018-08-01得票数 0
我正在开发一个数据流应用程序，我正在研究在这个项目中使用Apache的可能性。其主要原因是它支持很好的高级流结构，非常类似于Java 8的streaming。我将接收与数据库中特定记录相对应的事件，我希望能够处理这些事件(来自消息代理(如RabbitMQ或Kafka) )，并最终更新数据库中的记录，并将处理&#
浏览 1提问于2016-08-10得票数 13
回答已采纳
2回答
我正在通过spark structured向delta写入数据流。每个流批次包含key - value (还包含作为一列的时间戳)。delta lake不支持在源(蒸汽批)上使用多个相同的键进行更新，所以我只想用最新的时间戳记录来更新delta lake。我该怎么做呢？这是我正在尝试的代码片段： def upsertToDelta(microBatchOutputDF: DataFrame, bat
浏览 27提问于2020-06-19得票数 2
回答已采纳
我理解重新平衡可以在您的流上的任何时间和任何点发生。当它这样做时，由于没有为给定的偏移量提交最新的偏移量，可能会发生事件的重新处理。Kafka Streams是否允许在重新平衡发生之前完成任何动态处理？我的意思是，您的应用程序正在使用一个记录(在您的流程方法中)，一个重新平衡事件发生了。一个具体的</e
浏览 35提问于2020-01-14得票数 1
在我的本地汇合平台中，我有一个主题名为"FOO_02"，我手动向它插入了一些记录，因此，我可以根据以下命令从一开始就打印它：
我可以做这样的事情:我只想把COL1 = 1的记录拉出来吗？类似于我们可以使用where条件执行select语句来从普通数据库(如db2 )中提取数据。我尝试了以下命令，但是我相信它只会得到新的数据，因为我得到了这个命令的</e
浏览 1提问于2022-03-17得票数 0
回答已采纳
我的Kafka应用程序对每条消息做了一些转换逻辑，并将其转发到一个新的主题。应用程序中没有基于时间的聚合/处理，因此不需要使用任何自定义时间戳提取器。在从所有StreamThreads抛出这个异常(总共10个)之后，应用程序被冻结了，因为在流上有几个小时没有进一步的进展。在那之后也没有任何例外。当我重新启动应用程序时，它只开始处理新来的消息。现在的问题是，这些消息之间(抛
浏览 0提问于2016-12-27得票数 4
回答已采纳
我有一个从kafka主题中读取的spark流数据帧，我想在每次解析新记录时删除过去5分钟的重复数据。我知道dropDuplicates(["uid"])函数，但我不确定如何在特定的历史时间间隔内检查重复项。我的理解是： df = df.dropDuplicates(["uid"]) 要么处理当前(微)批处理读取的数据，要么处理</em
浏览 31提问于2020-04-21得票数 0
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
