Spark结构化流媒体应用程序中的死执行者答案

【问题标题】：dead executors in spark structured streaming appSpark结构化流媒体应用程序中的死执行者
【发布时间】：2021-02-09 09:53:50
【问题描述】：

我有一个简单的流式作业，它从 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 秒。我的 spark 配置如下。

        "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？

【问题讨论】：

标签： apache-spark spark-streaming spark-structured-streaming

【解决方案1】：

是的，当启用动态分配时，这是一个有效的场景。

在结构化流中，数据以微批量处理。如果 executor 空闲超时小于 microbatch 持续时间，则不断添加和删除 executor。但是，如果执行器空闲超时大于批处理持续时间，则永远不会删除执行器。控制此行为的属性是“spark.dynamicAllocation.executorIdleTimeout”，默认值为 60 秒。

因此，如果 60 秒内没有活动，则执行者将被移除。在您的情况下，由于触发间隔为 150 秒，因此 spark 相当快地处理 3-5k 事件的微批处理，并且执行程序有可能闲置超过 60 秒，因此被删除。

要更改此行为，请添加新配置“spark.dynamicAllocation.executorIdleTimeout”并将其设置为更高的值（例如 300 秒）。

【讨论】：