如何使用直播流API指定kafka spark流的消费组id。使用spark1.3
kafkaParams.put("group.id", "app1");
浏览 0提问于2016-04-09得票数 7
回答已采纳
1回答
应用服务器日志进程
、、、、
Clint希望从这些具有不同参数的数据中生成“访问”和“频率”报告。我最初的计划是使用kafka从应用服务器日志中获取这些数据-->使用kafka推送到Spark Streaming并处理这些数据-->将这些数据存储到HIVE -->使用zeppelin来获取那些经过处理和集中的日志数据但是据我所知，Kafka没有任何功能可以读取日志文件中的数据
浏览 31提问于2017-07-24得票数 0
试过：-从512到4096，较少失败，但即使是10s，失败仍然存在。.config("spark.streaming.kafka.maxRatePerPartition", "256")
.config("spark.streaming.
浏览 4提问于2017-02-16得票数 6
Spark从kafka获取数据。我想通过spark-streaming限制记录消耗。关于卡夫卡的数据非常庞大。我使用spark.streaming.kafka.maxRatePerPartition=1属性来限制spark中的记录。但是仍然在5分钟内，我收到了13400条消息。我的spark程序每5分钟不能处理超过1000条消息。Kafk
浏览 17提问于2018-02-06得票数 0
2回答
spark流中奇怪的延迟
、、、、
我最近一直在使用spark streaming来处理kafka中的数据。大多数情况下，数据处理在1-5秒内完成。然而，在几个批次之后，它连续花费了41 ~ 45秒，并且大部分延迟发生在从stage0获取数据的区域。我意外地发现Kafka的request.timemout.ms默认设置为40秒，并将此设置更改为10秒。实际处理时间
浏览 2提问于2017-01-18得票数 4
我确实看到我们需要在生产者端进行更改，以使用Gzip压缩，但我不确定如何在阅读消息时解压。请把灯熄灭，告诉我从哪里开始。我有我的端到端流工作的未压缩的消息。
浏览 0提问于2017-12-20得票数 2
我想处理从Kafka流到PySpark的日志数据并保存到拼图文件中，但我不知道如何将数据输入到Spark。请帮帮我谢谢。
浏览 68提问于2020-07-04得票数 0
我不知道为什么生产者发送的数据没有到达消费者手中。我正在使用cloudera虚拟机。我正在尝试编写简单的生产者消费者，其中生产者使用Kafka，消费者使用spark streaming。scala中的生产者代码：import org.apache.kafka.clients.producer._
object kafkaProducergenerator.getRandomV
浏览 0提问于2017-11-19得票数 1
2回答
使用Kafka(v2.11-0.10.1.0)-spark-streaming(v-2.0.1-bin-hadoop2.7).
我有Kafka生产者和Spark-streaming消费者来生产和消费。一切正常，直到我停止消费者(大约2分钟)并重新开始。消费者启动并读取数据，绝对完美。但是，我迷失了2分钟的数据，其中消费者是关闭的。, "", d
浏览 16提问于2017-01-18得票数 0
我提交了多个spark streaming作业，这些作业使用相同的group.id消耗同一个kafka topic，正如纯Kafka文档中所说的，拥有相同group.id的多个消费者将加入同一个消费组，并且kafka topic中分块的记录将被拆分成这些消费者。然而，正如我在我的工作中测试的那样，这两个spark streaming作业仍然使用相同的主题的所有分区(不是均匀分割的)，并且在who进程
浏览 63提问于2021-07-30得票数 0
2回答
实时事件处理
、、
我有一个事件源(比如油井中的传感器，大约50000个)，它向服务器产生事件。在服务器端，我想以这样的方式处理所有这些事件，即来自传感器的关于最新湿度、温度、压力...etc的信息将被存储/更新到数据库中。我真的需要spark，(flume/kafka) + spark，来满足处理端吗？
浏览 13提问于2017-06-29得票数 1
我的Spark作业抛出了一个异常，如下所示： at scala.Predef$.assert(Predef.scala:170)
我关闭了auto.commit (enable.auto.commit=false)，使用Kafka这个错误是Kafka消费者端的问题还是我的spark
浏览 2提问于2018-01-09得票数 0
我正在尝试创建一个spark consumer API来在我的消费者代码中接收来自Kafka.But的数据，我无法为这两个类添加jar/dependency :导入org.apache.spark.streaming.scheduler.ReceiverLauncher；导入org.apache.spark.streaming.Scheduler；
我在本地机器上使用Kafka 0.11.0.1和S
浏览 1提问于2017-09-27得票数 0
我在最新的文档页面中阅读了Spark Streaming kafka集成指南，它是基于Kafka 010版本的。group.id和Spark Streaming Direct Stream的关系是什么？
浏览 0提问于2016-10-13得票数 1
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
1回答
我正在使用spark structured streaming (2.2.1)来消费来自Kafka (0.10)的主题。val df = spark .format("kafka")
.option("kafka.bootstrap.servers", fromKafkaServers在某些情况下，我希望重新启动流应用程序并从头开始使用数据。然而，即使我
浏览 1提问于2018-04-18得票数 0
我正在从kafka流式传输数据，并试图将每批事件的数量限制在10个事件。在处理10-15批次后，批次大小突然出现峰值。下面是我的设置：
spark<
浏览 4提问于2017-11-17得票数 0
from channel, socket has likely been closed.at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:607)
at org.apache.spark.streaming.<
浏览 55提问于2017-07-07得票数 2
回答已采纳
我想知道，当流道是一个KafkaIO时，是否有人有过在Beam SparkRunner组件中限制速率的经验。我使用的版本是:Beam 2.29，Spark3.2.0和Kafka client 2.5.0？
我把光束参数maxRecordsPerBatch设置为一个很大的数，100000000。我的Kafka参数--fetchMaxWaitMs=1000 --maxPollRecord=5000应该能够带来大量的数据。特别是因为KafkaIO为每个分区创建了一个消费者。<
浏览 7提问于2022-03-14得票数 0
回答已采纳
1回答
使用Kafka Spark-Streaming。能够读取和处理生产者发来的数据。我在这里有一个场景，让我们假设生产者正在产生消息，而消费者被关闭了一段时间，然后打开了。> <groupId>org.apache.spark</groupId>
<artifactId>spark-streaming--
浏览 3提问于2017-02-06得票数 0
云服务器
ICP备案
云直播
即时通信 IM
实时音视频
腾讯云开发者
