我们在集成Spark-Kafka streams时遇到了性能问题。我们面临的问题是:在处理端，我们使用Spark直接流方法来处理相同的内容。如以下文档所示。Spark应该创建与主题中的分区数量相同的并行直接流(在本例中为3)。但是在读取时，我们可以看到来自分区1的所有消息都是首先处理的，然后是第二个，然后是第三个。有没有帮助解释为什么它不并行处理？根据我的理解，如果它同时从所有分区并行读取，那么消息输出应该是随机的。
浏览 3提问于2016-12-08得票数 5
例如，我们通常假设所有东西都运行在"foreachpartition“中，只有一个线程，我们不做线程安全锁，但是如果我们设置spark.streaming.concurrentJobs >1，我们应该注意线程安全吗
浏览 9提问于2020-06-05得票数 0
我在Spark Streaming Job中遇到了一种奇怪的行为。我们已使用spark.streaming.concurrentJobs的默认值1。
浏览 1提问于2017-04-18得票数 3
1回答
我用以下代码创建kafka流： KafkaUtils.createStream[....]( Map( .... ), StorageLevel.MEMORY_AND_DISK_SER .mapPartitions(.
浏览 0提问于2015-06-18得票数 3
spark.cassandra.connection.host", "127.0.0.1")
val conf = new SparkConf().setMaster("local[2]").setAppName("kafkaspark").set("spark.streaming.concurrentJobs
浏览 1提问于2017-05-09得票数 0
我的应用程序中运行着两个Spark流。在某些情况下，我需要查看第一个流是否创建了一个表，以便在第二个流中使用该表。我使用累加器作为指示器。因此，第一个流在完成其工作后更新此累加器的值，如果累加器的值已更改，则第二个流执行其逻辑--creates ABC Sql table --});
if(accumulator.value == 1){}到目前为止，
浏览 13提问于2018-01-12得票数 0
我正在尝试使用结构流从多个kafka主题中读取数据。版本kafka - 2.12-1.0.0 Spark Structure Streaming - 2.2.1val spark = SparkSession .appName("StreamLocallyExample") .config("spark.sql.streaming.checkpointLocation",
浏览 4提问于2018-04-19得票数 2
1回答
Kafka-火花流处理作业同步
、、、、
我正在尝试一个简单的测试，在那里我使用卡夫卡-连接和火花。SourceRecord sr = new SourceRecord(null, destTopic, cleanPath);val kafkaConsumerParams = Map[String, String](
浏览 0提问于2018-05-16得票数 1
回答已采纳
2回答
在这种情况下dStream.foreachRDD(a => ... )执行以下方法：
val dStream : Stream[_] = ...().cache()
dStream.toDb /
浏览 1提问于2016-05-24得票数 4
回答已采纳
2回答
纱线火花流:执行器未得到充分利用
、、、、
我和亚恩一起运行火花流-我正在通过DireactStream方法消费卡夫卡(没有接收者)。我有两个主题(每个有3个分区)。
我将RDD (我有一个DStream)分成16个部分(假设没有执行程序* num =2*8= 16 )，对吗？然后，我执行foreachPartition，并将每个分区写入本地文件，然后通过http (使用apach
浏览 3提问于2016-12-27得票数 2
我正在编写一个使用RabbitMQ来使用事件的星火流应用程序。我打算使用的RabbitMQ的一个特性是大量的消息，即不需要逐个对消息进行编码，而是只对批处理中的最后一个事件进行编码，这将对整个批处理进行攻击。这对于ack逻辑是至关重要的，因为如果RDD2可以在RDD1仍在处理时被处理，那么如果我在RDD2中的最后一个事件也会对RDD1中的所有事件进行编程，即使它们可能还没有被完全处理，也是如此。
浏览 0提问于2015-07-02得票数 1
回答已采纳
我有一个很大的kafka主题，我想从这个主题开始写批量偏移量。我目前正在使用以下方法。在本例中，stream是一个InputDStream (我最终在前面的DStream中使用了GenericRecord值)。 val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges val o: OffsetRange = offsetRanges(TaskC
浏览 0提问于2018-01-31得票数 0
1回答
我已经尝试过从4个不同的线程()发送这个作业多次(比如4次)，但是行为仍然相同。另外，我尝试将这个配置spark.streaming.concurrentJobs设置为超过1，但没有改变！
浏览 1提问于2019-10-03得票数 0
回答已采纳
我使用直接的Kafka API以1分钟的批处理间隔运行我的spark流媒体应用程序，并在我的应用程序代码中使用Pandas和Pyspark。我已经提供了如下参数的spark-submit作业，--deploy-mode cluster--total-executor-cores 4但是我的SPARK UI显示我的活动批处理处于队列状态，
config("spark.streaming.backp
浏览 14提问于2017-08-28得票数 1
2回答
我正在使用spark 1.5.2。我需要运行火花流工作与卡夫卡作为流媒体的来源。我需要阅读卡夫卡内部的多个主题，并对每个主题进行不同的处理。
浏览 4提问于2015-12-23得票数 21
回答已采纳
3回答
我的工作是火花流，我正面临一些问题，试图实现多个写。下面是我的代码DataWriter.writeStreamer(secondTableData,"parquet",CheckPointConf.secondCheckPoint,OutputConf.secondDataOutput)
DataWriter.writeStreamer(thir
浏览 2提问于2018-07-18得票数 1
回答已采纳
我们有一个在YARN集群上运行的Spark Streaming application。实际上，我们的处理时间超过了批处理间隔。Processing Time : 5 Minutes我们使用Direct Streaming approach从Kafka主题中获取数据。
我应该使用Window based操作吗？例如，如果我有Window length as 5 M
浏览 10提问于2017-02-07得票数 6
回答已采纳
1回答
可以在屏幕截图上观察到这些行为：虽然我知道根本原因可能很难找到并且不明显，但有没有方法可以进一步调试这个问题？目前我不知道从哪里开始挖掘。会不会和背压有关？truespark.streaming.backpressure.pid.minRate 1
spark.streaming.concurrentJobs
浏览 0提问于2020-09-24得票数 2
在spark中，您可以将spark.streaming.concurrentJobs设置为要运行的多个并发作业。
我想知道的是一种逻辑方法来确定我可以在给定环境中运行多少并发作业。
浏览 15提问于2017-02-23得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
