kafka直播流和seekToEnd

文章/答案/技术大牛

发布

1回答

scala、apache-spark、apache-kafka、spark-streaming-kafka

在我的Spark工作中，我用KafkaUtils.createDirectStream初始化Kafka流。我读到过Consumer的seekToEnd方法。如何将其应用于流？

浏览 12提问于2020-03-11得票数 1

1回答

kafka streams应用程序-重启时忽略旧消息

apache-kafka、apache-kafka-streams

我只想处理流应用程序启动后收到的数据，而不是之前提交的偏移量。重启后忽略kafka流app旧记录的正确方式是什么？对于kafka consumer API，我通常使用seekToEnd()方法来跳转到最新的记录。是否有用于流的等效机制？我希望避免过滤自上次提交以来的所有消息，以忽略旧消息。

浏览 3提问于2017-11-15得票数 2

1回答

Logstash -用于将数据保存在内存中的输入文件插件

ruby、elasticsearch、join、apache-kafka、logstash

我有1-一个单一的CSV文件和2-一个直播的KAFKA流。KAFKA stream引入了直播流日志，CSV文件包含元数据记录，我需要在将它们发送到Elastic Search之前将它们与流日志连接起来。Kafka流日志和CSV记录示例： KAFKA log: MachineID: 2424, MachineType: 1, MessageType: 9 CSV record: Ma

浏览 24提问于2020-11-01得票数 0

回答已采纳

1回答

为什么对卡夫卡的seekToBeginning和seekToEnd API的调用永远挂起？

java、apache-kafka

我的环境： kafkaConsumer.seekToBeginning(assignedPartitions);case LATEST: kafkaConsumer.seekToEnd它永远挂在seekToBeginning和seekToEnd调用上

浏览 3提问于2017-02-02得票数 3

1回答

如何使用c#消费最新的文本文件数据，用kafka发送并写回另一个文本文件？

c#、apache-kafka

我已经向Kafka producer发送了一个文本数据文件，在读取了该文件的字符串后。现在我想在文本文件中使用相同的数据。我如何消费它？var fileName = @"D:\kafka_examples\new2.txt";var consumer = new KafkaNet.Consumer(new ConsumerOp

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

3回答

：如何对分区进行seekToEnd？

spring、spring-boot、apache-kafka、spring-kafka

我使用Spring 2.0.2.RELEASE和Spring (有效的pom显示了spring的2.1.6.RELEASE版本)。final List<ConsumerRecord<String, EclLogging>> messages) throws Exception {public void seekToBeginning(String s, int i) { publ

浏览 0提问于2018-08-17得票数 0

回答已采纳

2回答

如何从read_committed Kafka消费者获取上次承诺的偏移量

apache-kafka、kafka-consumer-api、kafka-producer-api

我使用具有read_committed隔离级别的KafkaConsumer，并且我对seek和seekToEnd方法有问题。根据文档，seek和seekToEnd方法给出了LSO (最后一个稳定偏移量)。但这有点令人困惑。因为它总是给我相同的值，主题的结尾。无论最后一个条目是(由生产者)提交的还是已中止事务的一部分。但在消息期间，它会移动到主题的末尾(包括100_000 seekToEnd )。但是poll()不会返回它们。我正在寻找一种方法来检索上次提交的偏移量(因此生产者最后成功提交的消息)。但我假

浏览 3提问于2018-01-23得票数 6

3回答

如何在Spring Kafka客户端中查找偏移量？

java、apache-kafka、kafka-consumer-api、spring-kafka

在开发使用提要数据的微服务时，我们使用了ConsumerSeekAware和seekToBeginning。当我们准备好时，我们希望消费者将使用偏移量。由于文档的原因，我并不能真正理解seekToEnd代表什么。这是否意味着，它将查找到偏移量(最后一条ACKed消息)，还是将查找到主题的最后一条消息(无论偏移量如何)并等待新的消息？

浏览 0提问于2020-02-12得票数 0

1回答

卡夫卡-知道消费者是否是最新的

apache-kafka、kafka-consumer-api

我在本地Java消费者客户端上使用Kafka 0.9.0。如果我有一个主题和一个分区，有人能告诉我如果我这样做: seekToEnd(MyTopic)；poll(x)；我只会得到最后的记录，所以我会知道我在最后的位置？

浏览 0提问于2016-01-20得票数 1

回答已采纳

1回答

Spark kafka streaming -如何确定批次结束

apache-spark

我正在使用Kafka流媒体从Kafka主题消费。(kafka直播流)该topic中的数据每隔5分钟就会从其他来源到达。现在我需要处理每隔5分钟到达的数据，并将其转换为Spark DataFrame。现在，流是连续的数据流。我的问题是，我如何确定我已经读完了加载到Kafka主题中的第一组数据？

浏览 0提问于2017-06-09得票数 0

6回答

如何获取kafka主题分区的末尾偏移量？

java、apache-kafka、kafka-consumer-api

我正在使用Java编写一个kafka消费者。我想保持消息的实时，所以如果有太多的消息等待消费，比如1000或更多，我应该放弃未消费的消息，从最后一个偏移量开始消费。对于这个问题，我尝试比较上一个提交的偏移量和一个主题的结束偏移量(只有1个分区)，如果这两个偏移量之间的差异大于某个值，我会将该主题的最后一个提交的偏移量设置为下一个偏移量，这样我就可以丢弃那些冗余的消息

浏览 75提问于2016-07-18得票数 38

回答已采纳

1回答

Kafka Spark streaming中的阅读和处理并行性

scala、apache-spark、apache-kafka、spark-streaming、kafka-consumer-api

我正在尝试并行读取Kafka消息，从而并行处理它们。我的Kafka主题有10个分区。我正在尝试创建5个DStreams，并应用Union方法对单个DStream进行操作。scala.Array[Byte], String, DefaultDecoder, StringDecoder]( ssc, kafkaParams, Map("kafka_topic

浏览 0提问于2017-09-16得票数 0

1回答

使用opencv进行直播和处理

c++、opencv、image-processing、apache-kafka、video-streaming

我很难弄清楚一个看似简单的问题:我的目标是将视频流发送到服务器，使用opencv对其进行处理，然后将处理后的提要发回显示。我正在考虑使用kafka发送和接收feed，因为我已经有了一些经验。然而，这带来了一个问题: opencv使用VideoCapture方法处理视频流，这与仅使用Read方法读取单个图像不同。

浏览 0提问于2018-06-27得票数 0

1回答

从kafka获取所有消息，带开始和结束时间

apache-kafka

我在环境中有Kafka集群，我需要创建一个作业，根据开始和结束时间从Kafka主题的所有分区读取消息。所以，就像我想阅读从00:00:09:14:09:2020(SEC-MINUTES-HOURS-DAY-MONTH-YEAR)到00:00:10:14:09:2020 (持续1小时)的所有Kafka主题消息

浏览 67提问于2020-09-14得票数 0

1回答

如何使用kafka连接JDBC接收器和使用python的源

apache-kafka、apache-kafka-connect、kafka-python

我想要从一个系统到另一个系统的实时流。我正在使用kafka-python，并且能够在本地直播。计算出连接器将处理多个设备。有人能给我推荐一种在python中使用连接器实现它的方法吗？

浏览 28提问于2021-10-07得票数 1

回答已采纳

1回答

云直播和移动直播区别是什么？

云直播、移动直播

obs推流在手机上看直播怎么选择那一个？

浏览 1924提问于2019-01-11

2回答

Spark Streaming -在多个kafka流上加入操作速度较慢

scala、apache-spark、apache-kafka、spark-structured-streaming

我有3个kafka流，每个都有600k+记录，火花流需要超过10分钟来处理流之间的简单连接。Spark群集配置：这就是我在spark(scala)中读取kafka streams to tempviews的方法.format("kafka").option("sub

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

2回答

将最后一条信息发送给卡夫卡主题

apache-kafka、kafka-consumer-api、kafka-producer-api

我是卡夫卡的新手，我正在开发一个原型，将一个专有的流媒体服务连接到Kafka。 // Seek to the end of those partitions consumer.seekToEnd

浏览 5提问于2017-11-07得票数 3

回答已采纳

2回答

如何使用direct stream在Kafka Spark Streaming中指定消费群

java、apache-spark、apache-kafka、spark-streaming、kafka-consumer-api

如何使用直播流API指定kafka spark流的消费组id。

浏览 0提问于2016-04-09得票数 7

回答已采纳

2回答

为什么Kafka的seekToBeginning和seekToEnd不能与assign一起工作？

java、apache-kafka、kafka-consumer-api

比方说，我想检查Kafka中第一条和最后一条消息的偏移量，用于特定的分区。我的想法是将assign(…)方法与seekToBeginning(…)和seekToEnd(…)一起使用。如果将AUTO_OFFSET_RESET_CONFIG设置为"latest"，则seekToBeginning(…)无效；如果将其设置为"earliest"，则seekToEnd(…)不起作用。问题是:是否可以将seekToBeginning()或seekToEnd()

浏览 2提问于2019-10-25得票数 7

点击加载更多