最近，我将我们的流媒体应用程序从spark-streaming 2.1切换到使用kafka-streaming新API (1.0)和kafka broker服务器0.11.0.0
我已经实现了自己的Processor类，并且在process方法中，我只打印消息内容。我的问题是，在某些情况下，我在消息到达kafka主题/分区时就收到了消息，而在其他情况下，我在消息到达主题后10-15分钟才收
浏览 17提问于2018-02-14得票数 1
回答已采纳
1回答
我正在尝试创建一个简单的Flink Kafka消费者16:47:28,451 INFO org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09- Got 1 partitions from th
浏览 11提问于2019-07-13得票数 1
1回答
将消费者作为同一个消费者组的一部分意味着提供来自主题分区的消息分布在组成员之间的“竞争消费者”模式。: Previously tracked partitions [cpq.cluster-1] been revoked by Kafka because of consumer rebalance.Fundamentally different stream should use different group id
at org.apache.spark.streaming</
浏览 1提问于2019-10-14得票数 2
我在Scala中使用Spark在Kafka消费者应用程序中消费和处理消息。有时候，处理Kafka消息队列中的消息会比平时多花一点时间。此时，我需要消费最新的消息，忽略生产者已经发布但尚未消费的早期消息。properties = readProperties()
val streamConf = new SparkConf().setMaster("local[*]").se
浏览 3提问于2017-08-31得票数 1
2回答
我正在使用spark streaming从kafka读取数据。
为什么总任务量越来越大？经过几天的运行，恐怕会超过20亿……也许计数器会溢出？你的spark流媒体工作也有这种现象吗？
浏览 0提问于2015-05-13得票数 2
如何使用直播流API指定kafka spark流的消费组id。
浏览 0提问于2016-04-09得票数 7
回答已采纳
2回答
实时事件处理
、、
我把flume和kafka搞混了。我不想将事件存储在某个地方，因为我已经用最新的值更新了数据库。我真的需要spark，(flume/kafka) + spark，来满足处理端吗？
我们可以在没有水槽的情况下使用flume进行任何类型的处理吗？
浏览 13提问于2017-06-29得票数 1
我确实看到我们需要在生产者端进行更改，以使用Gzip压缩，但我不确定如何在阅读消息时解压。请把灯熄灭，告诉我从哪里开始。我有我的端到端流工作的未压缩的消息。
谢谢
浏览 0提问于2017-12-20得票数 2
1回答
我正在尝试读取xml文件并将其发送到kafka主题。这是我的卡夫卡代码，它把数据发送给卡夫卡-控制台-消费者。import org.apache.spark.SparkConf;import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaStreamingConte
浏览 0提问于2016-09-19得票数 2
1回答
;import org.apache.spark.streaming.api.java.JavaPairInputDStream;import org.apache.spark.streaming.kafka.KafkaUtils;
import or
浏览 3提问于2017-09-28得票数 1
回答已采纳
当我们准备好时，我们希望消费者将使用偏移量。由于文档的原因，我并不能真正理解seekToEnd代表什么。这是否意味着，它将查找到偏移量(最后一条ACKed消息)，还是将查找到主题的最后一条消息(无论偏移量如何)并等待新的消息？TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
log.info("Incoming Topic Partitions
浏览 0提问于2020-02-12得票数 0
我正在消费数据从卡夫卡主题通过火花结构化流，该主题有3个分区。由于Spark结构化流不允许您显式地提供group.id并向使用者分配一些随机id，所以我尝试检查使用者组id是否使用了kafka命令以下是我的问题
( 1)为甚麽要成立3个消费者团体？3)尽管我的spark应用程序仍在运行，但过了一段时间，这些组名没有出现在“消费者组”列表中。这是因为所有的数据都是由星火应用程序消耗的，而卡夫卡的主题中没有更多的数据吗？4)如果我的假设对第3点是正确的，那么
浏览 2提问于2019-02-28得票数 3
回答已采纳
：kafka.topic.tracking=trackings4。我关闭消费1，kafka自动完成再平衡，
浏览 3提问于2017-04-14得票数 5
回答已采纳
2回答
我们想知道对于给定的Kafka主题，还没有被Kafka Consumer消费和确认的消息数量。
有没有办法在Java中获得给定主题(主题有10个分区)中未消费的Kafka消息的计数?？
浏览 3提问于2019-10-24得票数 3
我想处理从Kafka流到PySpark的日志数据并保存到拼图文件中，但我不知道如何将数据输入到Spark。请帮帮我谢谢。
浏览 68提问于2020-07-04得票数 0
1回答
我正在用consumer.pause(<partitions>)实现卡夫卡延迟主题消费。Pub/Sub Kafka shim将暂停转换为NoOp：
是否有任何文件说明如何将酒吧次级主题的消费延迟一段时间？也就是说，我想使用来自Pub/Sub Lite主题的所有消息，但是要有4分钟的综合延迟。记录最近被轮询的recordsseparate记录转换为processrecords的足够老到processrecords还太小的
浏览 6提问于2021-11-17得票数 0
1:我们正在使用Spark Streaming进行接近实时的处理或批处理。我们目前的设计包含了Kafka。
2:生产者每隔15分钟发送一次消息。3:我们计划使用Spark Streaming来消费来自Kafka topic的消息。
浏览 2提问于2016-11-17得票数 2
我想从服务器获得从主题开始的所有消息。例如：
当使用上面的控制台命令时，我希望能够从一开始就获取主题中的所有消息，但是我无法使用java代码从一开始就使用主题中的所有消息。
浏览 4提问于2016-08-12得票数 13
回答已采纳
1回答
我用的是Kafka 0.8.1.1
有什么API (回调等)吗？使用哪个lost partitions或newly added partitions可以找到一个消费者？
浏览 2提问于2015-08-07得票数 2
1回答
我有一个主题，在Kafka broker中有3个分区，1个分区有消息，2个分区是空的，我如何知道在一次调用中消费哪个分区？首先，我分配了一个分区等于kafka.PartitionAny的TopicPartition，但是这个值一直返回-1for{
partitions = append(partitions, ka
浏览 0提问于2019-08-22得票数 0
云服务器
ICP备案
云直播
实时音视频
即时通信 IM
腾讯云开发者
