我有一个简单的流式处理任务，它从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秒）。