Spark Streaming -批处理间隔与处理时间
Spark Streaming是Apache Spark的一个组件，它提供了实时数据处理和流式计算的能力。它可以将实时数据流分成小批次，并在每个批次上执行批处理操作。
批处理间隔是指将实时数据流划分为批次的时间间隔。在Spark Streaming中，可以通过设置批处理间隔来控制数据流的处理频率。较短的批处理间隔可以实现更低的延迟，但也会增加处理的开销。较长的批处理间隔可以减少处理的开销，但会增加延迟。
处理时间是指每个批次的处理时间。它包括从接收数据到处理数据的整个过程，包括数据传输、转换、计算和输出等操作。处理时间的长短取决于数据量、计算复杂度和集群资源等因素。
Spark Streaming的优势在于其高吞吐量、低延迟和易于使用。它可以处理大规模的数据流，并提供了丰富的数据转换和计算操作，如过滤、映射、聚合和连接等。此外，Spark Streaming还与Spark的批处理引擎无缝集成，可以实现批处理和流式计算的混合应用。
Spark Streaming的应用场景包括实时数据分析、实时推荐、日志处理、网络监控和异常检测等。它可以处理各种类型的数据流，如传感器数据、日志数据、网络数据和社交媒体数据等。
腾讯云提供了一系列与Spark Streaming相关的产品和服务。其中，腾讯云的云服务器CVM提供了高性能的计算资源，可以用于部署Spark Streaming应用程序。腾讯云的对象存储COS可以用于存储和管理实时数据流。此外，腾讯云还提供了弹性MapReduce（EMR）和弹性容器实例（Elastic Container Instance）等服务，用于支持Spark Streaming的大规模计算和容器化部署。
更多关于腾讯云Spark Streaming相关产品和服务的信息，可以访问腾讯云官方网站：腾讯云Spark Streaming产品介绍。
相关·内容
在spark streaming中，我们接收的DStreams是一批RDDs。那么窗口如何进一步提供帮助呢？
据我所知，它还批处理RDDs。
如果我错了，请纠正我( Spark Streaming的新手)。
浏览 4提问于2015-10-08得票数 7
回答已采纳
我有一个spark-streaming服务，在那里我正在处理和检测基于某个离线生成的模型的异常。我从日志文件向此服务提供数据，日志文件使用以下命令进行流式传输
tail -f <logfile>| nc -lk 9999
在这里，spark流服务从端口9999获取数据。但是，我观察到最后几行正在被丢弃，即spark streaming没有接收到这些日志行，或者它们没有被处理。
但是，我也注意到，如果我只是将日志文件作为标准输入，而不是对其进行尾随，则不会删除任何行：
nc -q 10 -lk 9999 < logfile
有人能解释为什么会发生这种行为吗？对于将日志数据流式传
浏览 0提问于2016-02-09得票数 1
为了降低代码的难度，我允许重新启动Spark Streaming系统以使用新的批处理大小，但需要保留以前的进度(允许丢失正在处理的批处理)。 如果我在Spark Streaming中使用checkpoint，它不能在应用程序重新启动时更改所有配置。 所以我想通过修改源代码来实现这个功能，但是我不知道从哪里开始。希望能给我一些指导，告诉我其中的困难。
浏览 10提问于2020-07-19得票数 0
1回答
我使用SparkStreaming1.5.2，并使用Direct方法摄取Kafka 0.8.2.2中的数据。
我启用了检查点，这样我的驱动程序就可以重新启动，并在不丢失未经处理的数据的情况下拾取中断的位置。
检查点是写入S3的，因为我在Amazon上，而不是在Hadoop集群之上运行。
批处理间隔为1秒，因为我想要一个低延迟。
问题是，将一个检查点写入S3需要1到20秒。它们正在内存中备份，最终应用程序失败了。
2016-04-28 18:26:55,483 INFO [org.apache.spark.streaming.CheckpointWriter] [pool-16-thread-
浏览 0提问于2016-05-02得票数 4
我在Yarn上使用Spark Streaming，我面临着以下问题。
问题1：
我在yarn上使用spark streaming (1.6.1)，我总是看到活动作业计数为1，这意味着一次只有1个作业在运行。我已经使用了"--conf spark. streaming. concurrentJobs=3“参数，但不幸的是我总是只能看到1个活动作业。
问题2：
我有50个Kafka分区和spark streaming创建了50个RDD分区，但我可以看到95%的任务分配给了只有1个executor，其余的executor大多总是没有活动任务。
我的Spark Submit
浏览 0提问于2016-12-08得票数 0
我在Spark Streaming Job中遇到了一种奇怪的行为。我们已使用spark.streaming.concurrentJobs的默认值1。同一流式作业在批处理间隔设置为10分钟的情况下正常运行了一天多。突然，相同的作业开始对所有传入的批处理并发运行，而不是将它们放入队列中。
以前有没有人遇到过这种情况？这将是非常有帮助的！
浏览 1提问于2017-04-18得票数 3
我需要为我的应用程序增加每个分区的输入速率，并且我已经为配置使用了.set("spark.streaming.kafka.maxRatePerPartition",100)。流持续时间为10s，因此我希望处理此批的5*100*10=5000消息。然而，我收到的输入率只有500左右。你能建议任何修改来提高这个比率吗？
浏览 0提问于2016-12-08得票数 6
回答已采纳
我正在尝试从Kafka读取JSON消息，并使用spark structured将它们存储在HDFS中。
我遵循了示例，当我的代码如下所示时：
df = spark \
.read \
.format("kafka") \
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
.option("subscribe", "topic1") \
.load()
df.selectExpr("CAST(key AS STR
浏览 43提问于2018-07-25得票数 1
1回答
我有一个简单的流媒体工作，它从卡夫卡主题中提取数据并将其推送到S3。
df2 = parsed_df \
.coalesce(1)\
.writeStream.format("parquet")\
.option("checkpointLocation", "<s3location>")\
.option("path","s3location")\
.partitionBy("dt")\
.o
浏览 8提问于2020-10-27得票数 2
回答已采纳
我们每隔1分钟获取一些文本文件，并使用Spark Streaming将其聚合
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(60);
JavaDStream<String> file = ssc.textFileStream(inputDir)
但是，在聚合完成后，我们希望将聚合的JavaPairDStream<>与另一个提要连接起来，该提要是一个参考数据，每隔1小时到达一次。
在Spark Streaming中，是否可以加入在不同时间间隔到达的2个fee
浏览 1提问于2015-07-28得票数 0
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。
write_kafka_data.py:
read_df = spark.sql("select * from db.table where some_column in ('ASIA', 'Europe')")
final_df = read_df.select(F.to_json(F.struct(F.col("*"))).alias("value"))
final_df.write.format("kafka")\
浏览 0提问于2021-09-13得票数 4
回答已采纳
1回答
火花流待批处理
、、
我正在运行一个Spark Streaming应用程序，它从Kafka (使用Direct Stream方法)读取数据，并将结果发布回Kafka。该应用程序的输入速率以及应用程序的吞吐量在大约一两个小时内保持稳定。之后，我开始看到在Active Batches队列中保留很长时间(对于30mins+)的批处理。Spark driver日志指示以下两种类型的错误，并且这些错误发生的时间与批处理的开始时间非常吻合：
第一错误类型
ERROR LiveListenerBus: Dropping SparkListenerEvent because no remaining room in event q
浏览 3提问于2017-08-04得票数 0
1回答
我试图在每个RDD中添加max和min，在它的元组的星星之火dstream..each中。我编写了以下代码，但无法理解如何传递参数min和max。有人能建议一种方法来做这个转变吗？我尝试了以下几点：
JavaPairDStream<Tuple2<Long, Integer>, Tuple3<Integer,Long,Long>> sortedtsStream = transformedMaxMintsStream.transformToPair(new Sort2());
class MinMax implements Function<JavaPa
浏览 1提问于2015-06-17得票数 1
1回答
如何避免火花流中批量排队
、、、
我有直接流的火花流，我使用下面的配置
批间隔60
spark.streaming.kafka.maxRatePerPartition 42
auto.offset.reset最早
当我使用最早的选项启动流批时，为了更快地使用Kafka的消息并减少延迟，我将spark.streaming.kafka.maxRatePerPartition保持为42。因此，它应该消耗42x60x60分区=每批151200条记录。
我有两个问题
我看到，最初的几批记录正确地消耗了151200张唱片，尽管卡夫卡有大量的唱片可供消费，但在后来的批次中逐渐减少了。请看下面的截图。可能是什么原因
我看到很多
浏览 2提问于2021-03-16得票数 1
回答已采纳
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
我是spark streaming的新手，我需要一些基本的澄清，因为我不能完全理解阅读文档。
用例是，我有一组包含转储事件的文件，每个事件都有一个字段时间戳。
目前，我正在加载这个文件并提取JavaRDD中的所有事件，我想将它们传递给Spark Streaming，以便根据时间戳收集一些统计数据(类似于重放)。
我的问题是，是否可以使用事件时间戳作为时间参考来处理这些事件，而不是机器的实际时间(对于这个愚蠢的问题，很抱歉)。
在可能的情况下，我是需要简单的spark流，还是需要切换到结构化流？
我在这里发现了一个类似的问题：
提前感谢
浏览 1提问于2017-02-14得票数 2
1回答
火花结构流处理已处理过的故障记录
、、、、
我被困在火花结构流非常奇怪的问题。每当我关闭流并再次重新启动时，它将再次处理已处理的记录。
我尝试使用spark.conf.set("spark.streaming.stopGracefullyOnShutdown"，True)，但我仍然有问题。
任何关于如何解决这个问题的建议。
谢谢,
浏览 4提问于2022-03-29得票数 0
根据我们的实验，当状态变成超过一百万个对象时，有状态的星火流内部处理成本会花费大量的时间。因此，延迟会受到影响，因为我们必须增加批处理间隔以避免不稳定行为(处理时间>批处理间隔)。
它与我们的应用程序的细节无关，因为它可以通过下面的代码复制。
，那些花了这么多时间处理用户状态的星火内部处理/基础设施成本到底是多少？除了简单地增加批处理间隔之外，还有其他减少处理时间的选择吗？
我们计划广泛使用state :在少数几个节点中的每个节点上至少有100 it左右，以将所有数据保存在内存中，并且每小时只转储一次。
增加批处理间隔会有所帮助，但我们希望保持批处理间隔最小。
原因可能不是状态占用的空间
浏览 2提问于2015-09-10得票数 24
回答已采纳
我在yarn集群上使用spark-streaming (spark version 2.2)，并尝试为我的应用程序启用动态内核分配。 executor的数量可以根据需要增加，但是一旦分配了executor，即使流量减少了，它们也不会缩小，也就是说，一旦分配了executor就不会被释放。我还在yarn上启用了external shuffle服务，如下所述：https://spark.apache.org/docs/latest/running-on-yarn.html#configuring-the-external-shuffle-service 我在spark-submit命令中设置的配
浏览 19提问于2018-12-25得票数 1
回答已采纳
1回答
我正在使用Spark2.3.0、Scala2.11.8和Kafka，我试图用结构化流将来自Kafka的所有消息写入到parquet文件中，但是对于我的实现所做的每次查询，每次实现的总时间都会增加很多。我想知道为什么会发生这种情况，我尝试使用不同的可能触发器(继续，0秒，1秒，10秒，10秒，10分等等)，我总是得到相同的行为。我的代码有这样的结构：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Column, SparkSession}
import com.name.proto.ProtoMessag
浏览 1提问于2018-06-25得票数 2
回答已采纳
我正在尝试使用结构流从多个kafka主题中读取数据。版本kafka - 2.12-1.0.0 Spark Structure Streaming - 2.2.1
我的代码:-
val spark = SparkSession
.builder()
.appName("StreamLocallyExample")
.config("spark.master", "local")
.config("spark.sql.streaming.checkpointLocation",
浏览 4提问于2018-04-19得票数 2
Kafka已经填充了1个crore的消息，然后我执行这个代码库，按照预期它应该给出在2秒窗口中处理的元组/消息的计数，但它在第一次返回完整的消息计数，即1 crore，而且它给出0,0,0 ...Although打印该消息需要超过2秒。
内联源代码是
import java.io.Serializable;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import kafka.serializer.StringDecoder;
import
浏览 0提问于2015-05-28得票数 0
如果我是正确的，默认情况下，星火流1.6.1使用单个线程从每个Kafka分区读取数据，假设我的Kafka主题分区为50，这意味着每个50个分区中的消息将按顺序读取或可能以循环方式读取。
案例1：
-If是的，那么如何在分区级别并行读取操作？创建多个KafkaUtils.createDirectStream是唯一的解决方案吗？
e.g.
val stream1 = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSe
浏览 4提问于2016-12-12得票数 2
我使用直接的Kafka API以1分钟的批处理间隔运行我的spark流媒体应用程序，并在我的应用程序代码中使用Pandas和Pyspark。
下面是我的集群配置:3个数据节点，每个数据节点具有8核、12 my的容量。
我已经提供了如下参数的spark-submit作业，
--master yarn
--deploy-mode cluster
--executor-memory 2G
--total-executor-cores 4
--num-executors 11
但是我的SPARK UI显示我的活动批处理处于队列状态，
config("spark.streaming.backp
浏览 14提问于2017-08-28得票数 1
我写了一个有状态字计数火花流应用程序，可以从卡夫卡连续接收数据。我的代码包括一个mapWithState函数，可以正确运行。当我检查spark上的流统计时，我在处理时间图中发现了一些周期脉冲。我认为这可能是由于使用检查点造成的。希望有人能解释这件事，非常感谢！
和完成的分批表：
我发现一些1秒的成本成批是周期性的。然后，我进入一个1秒的时间成本批处理和一个亚秒时间成本的批处理，并发现1秒的时间成本批有多一个工作比另一个。
比较两种批次：
这似乎是由checkpoint引起的，但我不确定。
有人能为我详细解释一下吗？谢谢!
这是我的代码：
import kafka.s
浏览 1提问于2016-12-05得票数 1
回答已采纳
我的理解是Spark structured Streaming是建立在Spark SQL之上的，而不是Spark streaming。因此，下面的问题是，应用于spark流的属性是否也适用于spark结构化流，例如：
spark.streaming.backpressure.initialRate spark.streaming.backpressure.enabled spark.streaming.receiver.maxRate
浏览 1提问于2018-09-03得票数 1
我将一行从createStream改为createDirectStream，因为新库不支持createStream
我已经从这里检查过了，
scala> val lines = KafkaUtils.createDirectStream(ssc, "localhost:2181", "sparkgroup", topicpMap).map(_._2)
<console>:44: error: overloaded method value createDirectStream with alternatives:
[K, V
浏览 6提问于2022-05-15得票数 1
1回答
我在Spark Streaming 2.2.1工作中有16个接收器。过了一段时间，一些接收器处理的记录越来越少，最终每秒只能处理一条记录。可以在屏幕截图上观察到这些行为：
虽然我知道根本原因可能很难找到并且不明显，但有没有方法可以进一步调试这个问题？目前我不知道从哪里开始挖掘。会不会和背压有关？
Spark流属性：
spark.app.id application_1599135282140_1222
spark.cores.max 64
spark.driver.cores 4
spark.driver.extraJavaOptions -XX:+PrintFlagsFin
浏览 0提问于2020-09-24得票数 2
我们讨论了以下问题：
但是Spark Structured Streaming是在Spark2.2上添加的，它给流媒体带来了很多变化，而且它是非常出色的。
我们可以说Spark Strutured Streaming是流处理，还是仍然是批处理？
现在Apache Flink和Apache Spark Structured Streaming的最大区别是什么？
浏览 3提问于2017-09-01得票数 13
我集成了Spark Streaming Process和Kafka来阅读特定的主题。创建了Spark上下文，轮询时间为5秒，运行正常。但是，如果我想要实时访问提要，我可以将其进一步减少到1秒(会超过kill吗?)或者有没有其他更好的选择来处理这种情况。
浏览 0提问于2018-05-05得票数 0
我正试图将我的字数结果保存在文件中。
val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
wordCounts.saveAsTextFile("/home/hadoop/datafile1")
但它显示了
value saveAsTextFile is not a member of org.apache.spark.streaming.dstream.DStream[(String, Long)] [error] wordCounts.saveAsTextFile(&#
浏览 5提问于2017-04-14得票数 2
回答已采纳
1回答
我使用结构化流从Kafka读取数据，并创建各种聚合度量。我已经使用metrics.properties启用了石墨接收器。我已经看到应用程序在较早的星火版本有流相关的指标。我没有看到与结构化流相关的流标准。我做错了什么？
例如，无法找到未处理的批处理或正在运行的批处理或最后完成的批处理总延迟。
我通过设置以下方法启用了流度量：
SparkSession.builder().config("spark.sql.streaming.metricsEnabled",true)
即使如此，我也只能得到三个指标：
driver.spark.streaming.inputrate
d
浏览 4提问于2017-12-19得票数 12
3回答
我有一个spark流作业，它每5秒从Kafka读取一次，对传入的数据进行一些转换，然后写入文件系统。
这真的不需要是一个流作业，实际上，我只想每天运行一次，将消息排出到文件系统中。不过，我不确定如何停止这项工作。
如果我向streamingContext.awaitTermination传递一个超时，它不会停止进程，它所做的只是在迭代流时导致进程产生错误(参见下面的错误)。
完成我正在尝试做的事情的最佳方法是什么
这是针对Python上的Spark 1.6的
编辑：
感谢@marios，解决方案是这样的：
ssc.start()
ssc.awaitTermination(10)
ssc.sto
浏览 2提问于2016-01-30得票数 18
回答已采纳
不确定标题是否适合我想要实现的目标，所以请耐心等待。
我将从定义我的用例开始：
许多(比如说数百万) IoT设备正在向我的Spark stream发送数据。这些设备每10秒发送一次当前温度级别。所有这些IoT设备的所有者都有能力定义预设规则，例如:如果温度> 50，则执行某些操作。
我正在尝试计算是否可以输出这些设备中有多少设备在一段时间内满足了这一条件。问题是规则是实时定义的，应该实时应用于Spark作业。
我该怎么做呢。Spark是适合这项工作的工具吗？
非常感谢
浏览 0提问于2017-06-21得票数 1
为了研究目的，我从互联网上借来的一段代码令人费解。这是代码：
import org.apache.spark.sql.SparkSession
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable
val spark = ...
val sc = spark.sparkContext
val ssc = new StreamingContext(spark.sparkContext, Seco
浏览 0提问于2019-01-04得票数 1
回答已采纳
因此，在我们的一个kafka主题中，有接近100 GB的数据。我们正在运行spark-structured来获取S3格式的数据
当数据达到10 S3时，流可以正常运行，并且我们能够以S3格式获取数据。但由于有100 it的内存，在kafka中传输数据需要花费很长时间。
问: spark-streaming如何从Kafka读取数据？是否从当前偏移量中获取整个数据？或者是成批接受一定大小的产品？
浏览 0提问于2018-12-20得票数 3
3回答
我们正在尝试构建一个容错火花流作业，这是我们遇到的一个问题。下面是我们的场景：
1) Start a spark streaming process that runs batches of 2 mins
2) We have checkpoint enabled. Also the streaming context is configured to either create a new context or build from checkpoint if one exists
3) After a particular batch completes, the
浏览 3提问于2017-05-29得票数 4
我想了解"spark.streaming.blockInterval"在星火流DirectAPI中扮演了什么角色，据我理解，"spark.streaming.blockInterval"用于计算分区，即#partitions = (receivers x* batchInterval) /blockInterval，但在DirectAPI中，火花流分区等于no。卡夫卡隔断。
"spark.streaming.blockInterval"是如何在DirectAPI中使用的？
浏览 0提问于2016-12-17得票数 0
回答已采纳
在Spark streaming中，在DStreams上实现自定义分区程序的推荐方式是什么？
我在批处理模式下使用了JavaPairRDD.partitionBy(分区程序)，但在JavaDStreamPairRDD上还没有找到对应的程序。
谢谢
浏览 2提问于2015-04-27得票数 3
1回答
我有多个星星之火的结构化流作业，我看到的通常行为是，只有在Kafka中有任何新的偏移时才触发新的批处理，而Kafka是用来创建流查询的源。
但是，当我运行示例(演示使用mapGroupsWithState的任意有状态操作)时，即使流源中没有新的数据，也会触发新的批处理。为何会出现这种情况，是否可以避免呢？
Update-1 i修改了上面的示例代码，并删除与状态相关的操作，比如更新/删除它。函数只输出零。但是，在netcat服务器上，每10秒仍会触发一次批处理，而没有任何新的数据。
import java.sql.Timestamp
import org.apache.spark.sql.Sp
浏览 2提问于2020-07-07得票数 2
回答已采纳
默认情况下，结构化流作业使用HDFSStateStoreProvide。使用HDFS存储的问题是它是不可伸缩的。当作业在交通繁忙时间从kafka获得更多数据时，由于以下错误而失败：
18/12/06 15:54:35 ERROR scheduler.TaskSetManager: Task 191 in stage 231.0 failed 4 times; aborting job
18/12/06 15:54:35 ERROR streaming.StreamExecution: Query eventQuery [id = 42051afe-b1bc-438d-8143-2d7e5def
浏览 1提问于2018-12-07得票数 1
我设置火花流时间间隔为5s。如果当前的5s接收了非常非常多的数据，而spark streaming不能在5s内完成，但下一批数据即将到来。
spark streaming是否会同时处理下一批数据？
我的意思是批处理会并行执行吗？
浏览 10提问于2016-07-02得票数 3
回答已采纳
显然在Spark streaming中没有对Cassandra接收器的内置支持。我在网上找到了这个例子，它基于ForEachWriter为Spark structured streaming实现了一个自定义的Cassandra接收器： https://dzone.com/articles/cassandra-sink-for-spark-structured-streaming 我知道我们需要创建一个ForeachWriter实现，它负责打开到接收器(Cassandra)的连接，写入数据并关闭连接。所以CassandraSinkForeach和CassandraDriver类是有意义的。 但
浏览 31提问于2019-02-24得票数 0
这有点架构性的问题。我需要设计一个使用Spark和Scala作为主要工具的应用程序。我想尽可能地减少人工干预。
我会收到一个压缩与多个文件具有不同的结构作为输入在固定的时间间隔，例如，每天。我需要用Spark来处理它。转换后，需要将数据移动到后端数据库。
想要了解我可以用来设计应用程序的最佳方式。处理压缩包的最好方法是什么？Spark Streaming是否可以被视为查看文件频率的一种选择？我还应该考虑哪些其他选项？
任何指导都将是非常有价值的。
浏览 5提问于2017-08-18得票数 0
我正在对twitter上的直播流数据做一些分析。我听说了Spark streaming。我想知道哪些最适合直播流数据的分析，因为我的数据从来源来得非常快。
浏览 0提问于2014-11-22得票数 2
为了避免延迟和加快进程，我在spark流中构建了线程池。主程序列表如下：
stream.foreachRDD(rdd=> {
rdd.foreachPartition { rddPartition => {
val client: Client = ESClient.getInstance.getClient
var num = Random.nextInt()
val threadPool: ExecutorService = Executors.newFixedThreadPool(5)
val confs = new Configurati
浏览 0提问于2017-11-12得票数 1
我有一个Spark集群，总共有17个executors。我已经将Spark 2.1与Kafka集成在一起，并从topic中读取数据，例如：
val df = spark
.readStream
.format("kafka")
.options("kafka.bootstrap.servers","localhost:9092")
.options("subscribe","test")
.load
现在我想知道，当我以集群模式提交我的spark应用程序时，将分配多少个执行者(总共17个)来监
浏览 0提问于2018-09-05得票数 2
我使用Spark2.0.2(在DSE / DataStaX企业版5.1中)来运行一些流媒体应用。
我的Spark streaming应用程序为每个微批处理调用了一些RDD.persist()，而RDD.unpersist()从未被调用过(到目前为止，我们依靠缓存空间的LRU功能来解除持久化)。
我想我会在Spark UI的"Storage“选项卡中看到持久化RDD的列表增长了很多。
然而，在Spark UI的"Storage“选项卡中，我只看到了一个非常有限的持久化RDD列表。假设最大持久化RDD为10 MB，每个=> 1.5 MB持久化RDD占用的空间为15 MB，这是
浏览 0提问于2019-07-05得票数 0
1回答
我正在使用DirectAPI在纱线上运行火花流(1.6.1)来读取卡夫卡主题的事件，该主题有50个分区，并在HDFS上编写。我有60秒的间歇时间。我收到了大约500,000条信息，这些信息是在60 Sec下处理的。
火花突然开始接收到1500万到2000万条信息，用了5-6分钟的时间来处理，时间间隔为60秒。我已经配置了"spark.streaming.concurrentJobs=4"。
因此，当批处理需要很长的时间处理火花，启动并发的4个活动任务来处理积压批时，由于批处理间隔对于这样的数据量太小，批处理积压仍会在一段时间内增加。
我对此几乎没有疑问。
当我开始接收150
浏览 2提问于2016-12-16得票数 6
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
