我对spark streaming有严重的性能问题。对于10秒的批处理间隔，程序大约需要2分钟。我尝试在没有来自kafka主题的消息的情况下进行调试。即使没有要消费/处理的消息，大多数转换也需要30秒以上的时间。即使在decodeMessagesDF中</e
浏览 12提问于2019-02-13得票数 1
我有火花流应用程序从Kafka读取消息使用火花直接流(非接收者)方法和处理每个分区的消息。在我的Kafka分区中，有时我们得到处理2000条消息需要20秒的消息，而对于相同的no，有些消息需要7-9秒。信息的传递。
在波动的情况下，我们打开背压设置如下。读取RDD中的一个分区的2000条<em
浏览 2提问于2017-01-20得票数 3
我正在设置spark.streaming.kafka.maxRatePerPartition : 100。它在正常情况下运转良好。当消息在卡夫卡队列中的分布并不均衡时，问题就出现了。正在发生的情况是，即使分区中没有消息，而其他分区有更多的消息(比maxRatePerPartition更多)，但是spark批处理正在以较少
浏览 2提问于2021-09-29得票数 0
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
为了实现这一点，我提供了一个硬编码的JSON模式作为StructType。我正在寻找一种在流过程中动态推断主题模式的好方法。这是我的代码：(这是Kotlin，而不是通常使用的Scala) .readStream() .option("kafka.bootstrap.serversProcessingTime("25 seconds"))
.start("
浏览 2提问于2020-10-21得票数 1
回答已采纳
我的火花流版本是2.0，kafka版本是0.10.0.1，火花流-kafka-0-10_2.11。我使用直接的方式获得卡夫卡的记录，我现在想限制我在一批中获得的信息的最大数量。星星之火中的消费者数量是卡夫卡中的分区数?那么火花流中记录的最大数量是max.poll.records*consumers？
浏览 1提问于2018-09-27得票数 2
如何在Spark streaming中获取当前批量时间戳(DStream)？在执行过程中，我需要当前的时间戳来验证输入数据中的时间戳。如果我与当前时间进行比较，那么每次RDD转换执行时，时间戳可能会有所不同。
有没有办法获得时间
浏览 4提问于2015-12-23得票数 4
1回答
试图将org.apache.spark.sql.sources.CreatableRelationProvider转换为org.apache.spark.sql.execution.streaming.Sink，只需实现调用createRelation(...)的addBatch(...)，但createRelation(...)中有一个df.rdd，这会导致以下错误：
org.apache.spark.sql.catalyst.analysis.UnsupportedOperation
浏览 0提问于2018-08-06得票数 1
1回答
如果spark streaming在10秒的批处理间隔中获得50行消息，并且在40.5行消息之后10秒结束，而其余的消息落入另一个10秒间隔，则前40.5行文本是一个RDD首先被处理，在我的用例中，前40行是有意义的，但下一行.5没有意义，第二个RDD第一.5行也是如此，我的问题甚至是有效的?.
浏览 2提问于2016-03-11得票数 2
1回答
我正在执行Spark Streaming应用程序，并且正在缓存rdds用于历史回顾，我的批处理持续时间为1分钟，平均处理时间为14秒，因此执行器不会计算整个批处理持续时间。当我在内存中缓存rdd时，执行器也是如此。如果执行者被拖延，我们是否应该认为这种拖延执行者是浪费资源。
浏览 14提问于2017-06-29得票数 0
我确实看到我们需要在生产者端进行更改，以使用Gzip压缩，但我不确定如何在阅读消息时解压。请把灯熄灭，告诉我从哪里开始。我有我的端到端流工作的未压缩的消息。
谢谢
浏览 0提问于2017-12-20得票数 2
1回答
我正在尝试使用Spark Streaming将Kafka输入索引到elasticsearch中。kafka中的消息是这样的：我想在Spark Streaming中定义结构，以便在elasticsearch中索引此消息：我读过
浏览 1提问于2016-02-10得票数 1
我用pyspark写了一个Spark Streaming程序。
它通过socketTextStream接收实时输入的文本流，并进行相应的转换，然后通过saveAsTextFile将其保存为csv文件。不使用Spark streaming窗口操作，并且不需要以前的数据来创建输出数据。但是似乎在前一个RDD完成之前，Spark不会开始处理DStream中的RDD，即使
浏览 1提问于2015-04-29得票数 0
package com.scala.sparkStreaming
import org.apache.spark.streaming._
at org.apache.spark.streaming.StreamingContext.validate(StreamingContext.scala:517)
at org.apache.spark.st
浏览 2提问于2020-05-22得票数 1
我是spark streaming和elasticsearch的新手，我正在尝试使用spark从kafka topic读取数据，并将数据存储为rdd。在rdd中，我想在新数据到来时添加时间戳，然后推送到elasticsearch。 if(!
浏览 2提问于2017-10-26得票数 3
我正在运行Spark Streaming作业(意味着数据不断被推送到kafka主题，并被Spark消费者持续读取)。我的Kafka输入数据主题的保留时间设置为60000 (1分钟)。但是，Input Topic在1分钟后不会清除消息。如果没有新数据添加到主题中，则大约需要1分26分钟的时间来清除。如果我连续添加数据两分钟，由于retention.ms设置为1分钟，我预计将清除一半的
浏览 0提问于2015-05-01得票数 0
在Spark中有没有一种方法来处理超过水印的数据？ 考虑一个发送消息的设备的用例，这些消息需要在Kafka + Spark中处理。虽然99%的消息在10分钟内发送到Spark服务器，但偶尔设备可能会离开连接区一天或一周，并在内部缓冲消息，然后一旦连接恢复，就会在一周后发送。水印间隔必须相当有限，因为(1)主线情况下的结果必须及时
浏览 20提问于2020-08-26得票数 0
我目前正在对我的Spark流媒体应用程序进行性能测试。总而言之，我通过套接字连接接收JSON消息，每秒一条消息。然后，我使用Spark Streaming方法将消息作为RDDs处理，最后将生成的RDDs打印到数据库。这是我的问题:每当我想要检查块处理时间或调度延迟等时，我需要转到端口4040上的Spark UI。
我想做的是在程序运行时使用这些值
浏览 0提问于2016-05-06得票数 1
1回答
我使用的是火花流作业，它使用带有初始RDD的mapWithState。当重新启动应用程序并从检查点恢复时，它将失败，错误如下：
RDD转换和操作不是由驱动程序调用的，而是在其他转换中调用的；例如，rdd1.map(x => rdd2.values.count() * x)无效，因为值转换和计数操作不能在rdd1.map转换中执行。有关
浏览 12提问于2017-06-23得票数 12
7回答
我有Spark streaming应用程序，它基本上从Kafka获得触发消息，这启动了批处理，这可能需要2个小时。有一些事件，其中一些作业无限期地挂起，并且没有在通常的时间内完成，目前我们无法在不手动检查Spark UI的情况下确定作业的状态。我想知道当前正在运行的spark作业是否挂起的方法。我知道我可以使用驱动程序和执行器的度量标准。如果我要
浏览 6提问于2018-10-18得票数 8
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
