Spark Structured Streaming - stderr 被填满答案

【问题标题】：Spark Structured Streaming - stderr getting filled upSpark Structured Streaming - stderr 被填满
【发布时间】：2022-10-01 09:11:39
【问题描述】：

我在 GCP Dataproc 上有一个 Spark Structured Streaming 作业——它从 Kafka 中获取数据，进行处理并将数据推送回 kafka 主题。

几个问题：

Spark 是否将所有日志（包括 INFO、WARN 等）放入 stderr ？我注意到 stdout 是空的，而所有日志记录都放入 stderr

有没有办法让我使 stderr 中的数据过期（即使旧日志过期）？由于我有一个长时间运行的流式作业，stderr 会随着时间的推移而被填满，节点/虚拟机变得不可用。

请建议。

这是纱线日志命令的输出：

root@versa-structured-stream-v1-w-1:/home/karanalang# yarn logs -applicationId application_1663623368960_0008 -log_files stderr -size -500
2022-09-19 23:25:34,876 INFO client.RMProxy: Connecting to ResourceManager at versa-structured-stream-v1-m/10.142.0.62:8032
2022-09-19 23:25:35,144 INFO client.AHSProxy: Connecting to Application History server at versa-structured-stream-v1-m/10.142.0.62:10200
Can not find any log file matching the pattern: [stderr] for the container: container_e01_1663623368960_0008_01_000003 within the application: application_1663623368960_0008
Container: container_e01_1663623368960_0008_01_000002 on versa-structured-stream-v1-w-2.c.versa-sml-googl.internal:8026
LogAggregationType: LOCAL
=======================================================================================================================
LogType:stderr
LogLastModifiedTime:Mon Sep 19 23:25:35 +0000 2022
LogLength:43251469683
LogContents:
 applianceName=usa-isn0784-rt01, tenantName=NOV, mstatsTimeBlock=1663507200, tenantId=2, vsnId=0, mstatsTotSentOctets=11596, mstatsTotRecvdOctets=24481, mstatsTotSessDuration=300000, mstatsTotSessCount=1, mstatsType=sdwan-acc-ckt-app-stats, appId=https, site=usa-isn0784-rt01, accCkt=WAN-DIA, siteId=442, accCktId=1, user=10.126.117.196, risk=3, productivity=3, family=general-internet, subFamily=web, bzTag=Unknown,topic=syslog.ueba-us4.v1.versa.demo3,customer=versa  type(row) is ->  <class \'str\'>
End of LogType:stderr.This log file belongs to a running container (container_e01_1663623368960_0008_01_000002) and so may not be complete.
***********************************************************************


Container: container_e01_1663623368960_0008_01_000001 on versa-structured-stream-v1-w-1.c.versa-sml-googl.internal:8026
LogAggregationType: LOCAL
=======================================================================================================================
LogType:stderr
LogLastModifiedTime:Mon Sep 19 22:54:55 +0000 2022
LogLength:17367929
LogContents:
on syslog.ueba-us4.v1.versa.demo3-2
22/09/19 22:52:52 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-0f984ad9-f663-4ce1-9ef1-349419f3e6ec-1714963016-executor-1, groupId=spark-kafka-source-0f984ad9-f663-4ce1-9ef1-349419f3e6ec-1714963016-executor] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-2 to offset 449568676.
22/09/19 22:54:55 ERROR org.apache.spark.executor.CoarseGrainedExecutorBackend: RECEIVED SIGNAL TERM
End of LogType:stderr.
***********************************************************************


root@versa-structured-stream-v1-w-1:/home/karanalang# yarn logs -applicationId application_1663623368960_0008 -log_files stderr -size -500
2022-09-19 23:26:01,439 INFO client.RMProxy: Connecting to ResourceManager at versa-structured-stream-v1-m/10.142.0.62:8032
2022-09-19 23:26:01,696 INFO client.AHSProxy: Connecting to Application History server at versa-structured-stream-v1-m/10.142.0.62:10200
Can not find any log file matching the pattern: [stderr] for the container: container_e01_1663623368960_0008_01_000003 within the application: application_1663623368960_0008
Container: container_e01_1663623368960_0008_01_000002 on versa-structured-stream-v1-w-2.c.versa-sml-googl.internal:8026
LogAggregationType: LOCAL
=======================================================================================================================
LogType:stderr
LogLastModifiedTime:Mon Sep 19 23:26:02 +0000 2022
LogLength:44309782124
LogContents:
, tenantId=3, vsnId=0, mstatsTotSentOctets=48210, mstatsTotRecvdOctets=242351, mstatsTotSessDuration=300000, mstatsTotSessCount=34, mstatsType=dest-stats, destIp=165.225.216.24, mstatsAttribs=,topic=syslog.ueba-us4.v1.versa.demo3,customer=versa  type(row) is ->  <class \'str\'>
22/09/19 23:26:02 WARN org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer: KafkaDataConsumer is not running in UninterruptibleThread. It may hang when KafkaDataConsumer\'s methods are interrupted because of KAFKA-1894
End of LogType:stderr.This log file belongs to a running container (container_e01_1663623368960_0008_01_000002) and so may not be complete.
***********************************************************************


Container: container_e01_1663623368960_0008_01_000001 on versa-structured-stream-v1-w-1.c.versa-sml-googl.internal:8026
LogAggregationType: LOCAL
=======================================================================================================================
LogType:stderr
LogLastModifiedTime:Mon Sep 19 22:54:55 +0000 2022
LogLength:17367929
LogContents:
on syslog.ueba-us4.v1.versa.demo3-2
22/09/19 22:52:52 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-0f984ad9-f663-4ce1-9ef1-349419f3e6ec-1714963016-executor-1, groupId=spark-kafka-source-0f984ad9-f663-4ce1-9ef1-349419f3e6ec-1714963016-executor] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-2 to offset 449568676.
22/09/19 22:54:55 ERROR org.apache.spark.executor.CoarseGrainedExecutorBackend: RECEIVED SIGNAL TERM
End of LogType:stderr.

标签： apache-spark pyspark spark-structured-streaming google-cloud-dataproc

【解决方案1】：

简短的回答：

您可以使用带有 RollingFileAppender 的自定义 log4j 配置来限制长时间运行的作业的日志大小。

长答案：

Dataproc 上 Spark 的默认 log4j 配置为 /etc/spark/conf/log4j.properties。它在 INFO 级别将根记录器配置为 stderr。但在运行时，驱动程序日志（在客户端模式下）将由 Dataproc 代理定向到 GCS 并流回客户端，而执行程序日志（和集群模式下的驱动程序日志）将由 YARN 重定向到 stderr 文件中的容器的 YARN 日志目录。考虑使用 /etc/spark/conf/log4j.properties 作为自定义配置的模板。

在您的自定义配置中，您可以配置要写入 RollingFileAppender 的日志，例如，

log4j.rootLogger=INFO, rolling_file

log4j.appender.rolling_file=org.apache.log4j.RollingFileAppender
log4j.appender.rolling_file.File=${spark.yarn.app.container.log.dir}/my_app.log
log4j.appender.rolling_file.MaxFileSize=100MB
log4j.appender.rolling_file.MaxBackupIndex=10
...

请注意，对于执行程序（和集群模式下的驱动程序），log4j.appender.rolling_file.File 的值需要是${spark.yarn.app.container.log.dir} 下的路径，请参阅此question 和此doc。

将您的 log4j 配置上传到 GCS 存储桶，驱动程序和执行程序可能共享也可能不共享相同的配置。在您的情况下，您可能只想更新执行程序 log4j 配置，只需使用驱动程序的默认值。

然后通过以下方式之一使用自定义 log4j 配置提交作业：

文件名必须是log4j.properties，驱动程序和执行程序将共享相同的配置：
```
gcloud dataproc jobs submit spark ... \
  --files gs://my-bucket/log4j.properties
```
1. 文件名不必是log4j.properties，驱动程序和执行程序可以有不同的配置：
```
gcloud dataproc jobs submit spark ... \
  --files gs://my-bucket/my-log4j.properties \
  --properties 'spark.executor.extraJavaOptions=-Dlog4j.configuration=file:my-log4j.properties'
```
期望在 YARN 容器日志目录下会有 Spark 执行器的滚动日志，它们将自动聚合并存储在 GCS 和 Cloud Logging 中。

【讨论】：

感谢您的详细说明，我会尽快检查并回复