[SPARK-43002][YARN] Modify yarn client application report logging frequency to reduce noise

### What changes were proposed in this pull request?
* Added a new config property — `spark.yarn.report.loggingFrequency`
* Limit the number of times the yarn application report is logged based on the number of reports processed.

### Why are the changes needed?
Currently, an application report is generated every second, this tends to add a lot of noise especially for long running applications. This bloats the log file and makes it hard to navigate. With this change, we can limit the amount of times the application status report is logged based on the number of reports processed.

### Does this PR introduce _any_ user-facing change?
The logs are now ~30s apart
```
31-03-2023 15:00:08 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:00:08 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: RUNNING)
31-03-2023 15:00:38 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:00:38 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: RUNNING)
31-03-2023 15:01:08 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:01:08 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: RUNNING)
31-03-2023 15:01:38 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:01:38 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: RUNNING)
31-03-2023 15:02:09 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:02:09 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: RUNNING)
31-03-2023 15:02:31 PDT countByCountryFlow_countByCountry INFO - 23/03/31 22:02:31 INFO yarn.Client: Application report for application_1676870052658_5870144 (state: FINISHED)
```

### How was this patch tested?
Tested locally to ensure the behavior was as expected

Closes #40637 from ShreyeshArangath/ShreyeshArangath/yarn-log-frequency-patch.

Authored-by: Shreyesh Shaju Arangath <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: Shreyesh Shaju Arangath

docs/running-on-yarn.md

@@ -646,6 +646,16 @@ To use a custom metrics.properties for the application master and executors, upd
   </td>
   <td>0.9.0</td>
 </tr>
+<tr>
+  <td><code>spark.yarn.report.loggingFrequency</code></td>
+  <td><code>30</code></td>
+  <td>
+    Maximum number of application reports processed until the next application status 
+    is logged. If there is a change of state, the application status will be logged regardless
+    of the number of application reports processed.
+  </td>
+  <td>3.5.0</td>
+</tr>
 <tr>
   <td><code>spark.yarn.clientLaunchMonitorInterval</code></td>
   <td><code>1s</code></td>

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -1142,6 +1142,8 @@ private[spark] class Client(
       logApplicationReport: Boolean = true,
       interval: Long = sparkConf.get(REPORT_INTERVAL)): YarnAppReport = {
     var lastState: YarnApplicationState = null
+    val reportsTillNextLog: Int = sparkConf.get(REPORT_LOG_FREQUENCY)
+    var reportsSinceLastLog: Int = 0
     while (true) {
       Thread.sleep(interval)
       val report: ApplicationReport =
@@ -1160,9 +1162,12 @@ private[spark] class Client(
               Some(msg))
         }
       val state = report.getYarnApplicationState
-
+      reportsSinceLastLog += 1
       if (logApplicationReport) {
-        logInfo(s"Application report for $appId (state: $state)")
+        if (lastState != state || reportsSinceLastLog >= reportsTillNextLog) {
+          logInfo(s"Application report for $appId (state: $state)")
+          reportsSinceLastLog = 0
+        }
 
         // If DEBUG is enabled, log report details every iteration
         // Otherwise, log them every time the application changes state

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala

@@ -225,6 +225,18 @@ package object config extends Logging {
     .timeConf(TimeUnit.MILLISECONDS)
     .createWithDefaultString("1s")
 
+  private[spark] val REPORT_LOG_FREQUENCY = {
+    ConfigBuilder("spark.yarn.report.loggingFrequency")
+      .doc("Maximum number of application reports processed " +
+        "until the next application status is logged. " +
+        "If there is a change of state, the application status will be logged " +
+        "regardless of the number of application reports processed.")
+      .version("3.5.0")
+      .intConf
+      .checkValue(_ > 0, "logging frequency should be positive")
+      .createWithDefault(30)
+  }
+
   private[spark] val CLIENT_LAUNCH_MONITOR_INTERVAL =
     ConfigBuilder("spark.yarn.clientLaunchMonitorInterval")
       .doc("Interval between requests for status the client mode AM when starting the app.")