我的火花流版本是2.0，kafka版本是0.10.0.1，火花流-kafka-0-10_2.11。我使用直接的方式获得卡夫卡的记录，我现在想限制我在一批中获得的信息的最大数量。星星之火中的消费者数量是卡夫卡中的分区数?那么火花流中记录的最大数量是max.poll.records*consumers？
浏览 1提问于2018-09-27得票数 2
我对spark结构化数据流和kafka的偏移管理是新的。使用spark-streaming-kafka-0-10-2.11。在消费者中，我如何从主题的特定分区中读取？comapany_df = sparkSession .format("kafka").option("kafka</
浏览 22提问于2019-05-29得票数 0
我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。我在Kafka和Spark两边都没有发现任何堆栈或线索。
有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？
浏览 0提问于2016-05-18得票数 0
如果我是正确的，默认情况下，星火流1.6.1使用单个线程从每个Kafka分区读取数据，假设我的Kafka主题分区为50，这意味着每个50个分区中的消息将按顺序读取或可能以循环方式读取。案例1：e.g.._2)-If --我的卡夫卡分区每秒
浏览 4提问于2016-12-12得票数 2
2回答
spark流中奇怪的延迟
、、、、
我最近一直在使用spark streaming来处理kafka中的数据。大多数情况下，数据处理在1-5秒内完成。然而，在几个批次之后，它连续花费了41 ~ 45秒，并且大部分延迟发生在从stage0获取数据的区域。我意外地发现Kafka的request.timemout.ms默认设置为40秒，并将此设置更改为10秒。实际处理时间为1-5秒。<em
浏览 2提问于2017-01-18得票数 4
我向Kafka发送消息，我希望在其他应用程序中通过spark streaming接收这些消息，但Spark接收的消息速率约为40,000。我想增加Spark每个间隔接收的消息数，我该怎么做呢？
浏览 0提问于2017-03-11得票数 1
我有一个独立的spark集群，正在从kafka队列中读取数据。kafka队列有5个分区，spark只处理其中一个分区的数据。> <version>2.0.2</versi
浏览 2提问于2017-02-27得票数 3
回答已采纳
2回答
我的星火流应用程序是从卡夫卡读取使用DStream方法，我试图获得批量大小，以处理60,000条消息在10秒内。我所做的，
在纱线模式下运行，有2个执行器(4个芯，3个
浏览 0提问于2019-07-08得票数 2
回答已采纳
我有火花流应用程序从Kafka读取消息使用火花直接流(非接收者)方法和处理每个分区的消息。在我的Kafka分区中，有时我们得到处理2000条消息需要20秒的消息，而对于相同的no，有些消息需要7-9秒。信息的传递。spark.streaming.kafka.maxRatePerPartition=200
spark</em
浏览 2提问于2017-01-20得票数 3
如果一个kakfa主题在java中有多个分区，那么这许多使用者实例/线程将在使用者端实例化。
如何在火花缭乱的消费端处理它呢？我找不到很多关于同样的信息。
浏览 1提问于2019-06-10得票数 0
我有一个Kafka分区和一个sparkStreaming应用程序。一台具有10核的服务器。当spark streaming从Kafka得到一条消息时，后续过程将需要5秒(这是我的代码)。所以我发现sparkStreaming读取Kafka消息非常慢，我猜当spark读出一条消息时，它会一直等到消息被处理，所以读取和处理是同步的。我想知道我可以让spark异步读取吗？因此，Kafka
浏览 0提问于2015-09-06得票数 1
如何使用直播流API指定kafka spark流的消费组id。使用spark1.3
kafkaParams.put("group.id", "app1");
浏览 0提问于2016-04-09得票数 7
回答已采纳
Spark从kafka获取数据。我想通过spark-streaming限制记录消耗。关于卡夫卡的数据非常庞大。我使用spark.streaming.kafka.maxRatePerPartition=1属性来限制spark中的记录。但是仍然在5分钟内，我收到了13400条消息。我的spark程序每5分钟不能处理超过1000条消息。Kafka topic有3个分区。
浏览 17提问于2018-02-06得票数 0
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
3回答
当我从Kafka主题创建一个流并打印它的内容时 os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell' from pyspark.streamingimpo
浏览 2提问于2019-12-08得票数 1
回答已采纳
我使用的是Spark流媒体应用程序。应用程序使用直接流从Kafka topic (具有200个分区)中读取消息。:1361)at org.
浏览 1提问于2017-12-03得票数 2
回答已采纳
我们在集成Spark-Kafka streams时遇到了性能问题。项目设置:我们使用3个分区的Kafka主题，每个分区产生3000条消息，并在Spark direct streaming中进行处理。我们面临的问题是:在处理端，我们使用Spark直接流方法来处理相同的内容。如以下文档所示。Spark应该创建与主题中的分区数量相同的并行直接流(在本例中为3)。但是在读取时，我们可以看到来自分区<
浏览 3提问于2016-12-08得票数 5
1回答
应用服务器日志进程
、、、、
Clint希望从这些具有不同参数的数据中生成“访问”和“频率”报告。我最初的计划是使用kafka从应用服务器日志中获取这些数据-->使用kafka推送到Spark Streaming并处理这些数据-->将这些数据存储到HIVE -->使用zeppelin来获取那些经过处理和集中的日志数据但是据我所知，Kafka没有任何功能可以读取日志文件中的数据，并将它们发布在<
浏览 31提问于2017-07-24得票数 0
我有我的spark应用程序，从Kafka读取数据并摄取到Kudu。它已经成功运行了近25个小时，并将数据摄入到了Kudu中。在那之后，我看到从kafka日志中选出了新的kafka分区的领导者。我的应用程序进入完成状态，并出现以下错误
org.apache.spark.SparkException: ArrayBuffer(kafka.common.NotLeaderForPartitionException, org
浏览 20提问于2018-02-13得票数 0
在我的Spark Streaming作业中，CPU未得到充分利用(仅5-10%)。
它从Kafka获取数据并发送到DynomoDB或thridparty端点。
浏览 1提问于2017-03-18得票数 0
云服务器
ICP备案
云直播
实时音视频
即时通信 IM
腾讯云开发者
