我有一个简单的流式处理任务,它从kafka主题中提取数据并将其推送到s3。
df2 = parsed_df \
.coalesce(1)\
.writeStream.format("parquet")\
.option("checkpointLocation", "<s3location>")\
.option("path","s3location")\
.partitionBy("dt")\
.outputMode("Append")\
.trigger(processingTime='150 seconds')\
.start()
触发时间为150秒。我的星火配置是下面这个工作。
"driverMemory": "6G",
"driverCores": 1,
"executorCores": 1,
"executorMemory": "3G",
{
"spark.dynamicAllocation.initialExecutors": "3",
"spark.dynamicAllocation.maxExecutors": "12",
"spark.driver.maxResultSize": "4g",
"spark.sql.session.timeZone":"UTC",
"spark.executor.memoryOverhead": "1g",
"spark.driver.memoryOverhead": "2g",
"spark.dynamicAllocation.enabled": "true",
"spark.rpc.message.maxSize": "1024",
"spark.streaming.receiver.maxRate": "4000",
"spark.port.maxRetries" : "100",
"spark.jars.packages": "org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.4,org.apache.spark:spark-streaming-kafka-0-10-assembly_2.12:2.4.4"
}
工作进展顺利。但当我检查我的spark ui时,我看到很多死了的执行者。
这些死去的遗嘱执行人不断增加。每一批150秒,我处理3-5k事件。我的问题是:-
这是一个有效的方案吗?
如果这不是一个有效的场景,那么原因是什么?是因为动态分配属性设置为true吗?
1条答案
按热度按时间nfg76nw01#
是的,当启用动态分配时,这是一个有效的场景。
在结构化流媒体中,数据在微批量中处理。如果执行器空闲超时小于微批处理持续时间,则会不断添加和删除执行器。但是,如果执行器空闲超时大于批处理持续时间,则永远不会删除执行器。控制此行为的属性是“spark.DynamicLocation.ExecuteIdleTimeout”,默认值为60秒。
因此,如果没有活动,执行者将被移除60秒。在您的案例中,由于触发间隔为150秒,因此spark会相当快地处理3-5k事件的微批处理,并且执行器有可能闲置超过60秒,因此被删除。
要更改此行为,请添加一个新配置“spark.dynamicalocation.executidletimeout”,并将其设置为更高的值(例如300秒)。