lean_ctx_patterns/
spark.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Apache Spark (`spark-submit`) log compression.
4//!
5//! Spark drivers emit hundreds of `YY/MM/DD HH:MM:SS INFO Component: ...`
6//! lines. We drop INFO noise, keep finished-job lines, deduplicate WARNs and
7//! preserve ERRORs / exceptions, then prefix a one-line job/warn/error count.
8//!
9//! Crucially we also keep lines that are NOT framework log records — these are
10//! the application's own stdout (e.g. `Result: total words = 184273`), which is
11//! the actual point of the run and must never be dropped.
12
13use crate::ansi::strip_ansi;
14use std::collections::HashSet;
15
16pub fn compress(_cmd: &str, output: &str) -> Option<String> {
17    let trimmed = output.trim();
18    if trimmed.is_empty() {
19        return Some("spark: ok".to_string());
20    }
21
22    let mut jobs: Vec<String> = Vec::new();
23    let mut warnings: Vec<String> = Vec::new();
24    let mut warn_seen: HashSet<String> = HashSet::new();
25    let mut errors: Vec<String> = Vec::new();
26    let mut app_output: Vec<String> = Vec::new();
27    let mut saw_log = false;
28
29    for raw in trimmed.lines() {
30        let stripped = strip_ansi(raw);
31        let line = stripped.trim();
32        if line.is_empty() {
33            continue;
34        }
35
36        if let Some((level, rest)) = parse_log(line) {
37            saw_log = true;
38            match level {
39                "INFO" => {
40                    if rest.contains("Job ") && rest.contains("finished") {
41                        jobs.push(rest.to_string());
42                    }
43                }
44                "WARN" => {
45                    if warn_seen.insert(normalize(rest)) {
46                        warnings.push(rest.to_string());
47                    }
48                }
49                "ERROR" => errors.push(rest.to_string()),
50                _ => {}
51            }
52        } else if is_exception(line) {
53            errors.push(line.to_string());
54        } else {
55            // Not a framework log line → application stdout. Preserve it.
56            app_output.push(line.to_string());
57        }
58    }
59
60    if !saw_log && jobs.is_empty() && errors.is_empty() {
61        return Some(fallback(trimmed));
62    }
63
64    let mut parts = vec![format!(
65        "spark: {} job(s), {} warning(s), {} error(s)",
66        jobs.len(),
67        warnings.len(),
68        errors.len()
69    )];
70    push_capped(&mut parts, &jobs, 10, "more jobs");
71    push_capped(&mut parts, &warnings, 5, "more warnings");
72    push_capped(&mut parts, &errors, 10, "more errors");
73    push_capped(&mut parts, &app_output, 20, "more output lines");
74    Some(parts.join("\n"))
75}
76
77/// Parse a `DATE TIME LEVEL rest` Spark log line into `(level, rest)`.
78fn parse_log(line: &str) -> Option<(&str, &str)> {
79    let mut it = line.splitn(4, ' ');
80    let date = it.next()?;
81    let time = it.next()?;
82    let level = it.next()?;
83    let rest = it.next().unwrap_or("");
84    if looks_like_date(date) && looks_like_time(time) && is_level(level) {
85        Some((level, rest))
86    } else {
87        None
88    }
89}
90
91fn looks_like_date(s: &str) -> bool {
92    let parts: Vec<&str> = s.split('/').collect();
93    parts.len() == 3 && parts.iter().all(|p| p.chars().all(|c| c.is_ascii_digit()))
94}
95
96fn looks_like_time(s: &str) -> bool {
97    let parts: Vec<&str> = s.split(':').collect();
98    parts.len() == 3 && parts.iter().all(|p| p.chars().all(|c| c.is_ascii_digit()))
99}
100
101fn is_level(s: &str) -> bool {
102    matches!(s, "INFO" | "WARN" | "ERROR" | "DEBUG" | "TRACE")
103}
104
105fn is_exception(line: &str) -> bool {
106    line.contains("Exception") || line.starts_with("Caused by:") || line.contains("Error:")
107}
108
109/// Collapse digits so "took 5.1 s" / "took 9.2 s" warnings dedupe together.
110fn normalize(s: &str) -> String {
111    s.chars().filter(|c| !c.is_ascii_digit()).collect()
112}
113
114fn push_capped(parts: &mut Vec<String>, items: &[String], cap: usize, label: &str) {
115    for item in items.iter().take(cap) {
116        parts.push(format!("  {item}"));
117    }
118    if items.len() > cap {
119        parts.push(format!("  ... +{} {label}", items.len() - cap));
120    }
121}
122
123fn fallback(text: &str) -> String {
124    let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
125    let n = lines.len().min(8);
126    let mut s = lines[..n].join("\n");
127    if lines.len() > n {
128        s.push_str(&format!("\n... (+{} lines)", lines.len() - n));
129    }
130    s
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136
137    const LOG: &str = "23/01/01 12:00:00 INFO SparkContext: Running Spark version 3.4.0\n23/01/01 12:00:01 INFO ResourceUtils: No custom resources configured\n23/01/01 12:00:02 INFO Utils: Successfully started service\n23/01/01 12:00:03 WARN NativeCodeLoader: Unable to load native-hadoop\n23/01/01 12:00:10 INFO DAGScheduler: Job 0 finished: collect at Main.scala:20, took 5.123 s\n23/01/01 12:00:15 ERROR Executor: Exception in task 0.0 in stage 1.0\n";
138
139    #[test]
140    fn drops_info_keeps_job_warn_error() {
141        let r = compress("spark-submit app.py", LOG).unwrap();
142        assert!(r.contains("1 job(s), 1 warning(s), 1 error(s)"), "{r}");
143        assert!(r.contains("Job 0 finished"), "{r}");
144        assert!(r.contains("Executor: Exception"), "{r}");
145        assert!(!r.contains("ResourceUtils"), "drops info noise: {r}");
146        assert!(!r.contains("23/01/01"), "drops timestamps: {r}");
147    }
148
149    #[test]
150    fn keeps_application_stdout() {
151        let log = "23/01/01 12:00:00 INFO SparkContext: Running Spark version 3.4.0\n23/01/01 12:00:01 INFO ResourceUtils: No custom resources\n23/01/01 12:00:10 INFO DAGScheduler: Job 0 finished: collect, took 5.1 s\nResult: total words = 184273\n23/01/01 12:00:11 INFO SparkContext: Successfully stopped";
152        let r = compress("spark-submit app.py", log).unwrap();
153        assert!(
154            r.contains("Result: total words = 184273"),
155            "keeps app output: {r}"
156        );
157        assert!(!r.contains("ResourceUtils"), "still drops info noise: {r}");
158    }
159
160    #[test]
161    fn shorter_than_input() {
162        let r = compress("spark-submit app.py", LOG).unwrap();
163        assert!(r.len() < LOG.len());
164    }
165
166    #[test]
167    fn empty_is_ok() {
168        assert_eq!(compress("spark-submit app.py", "").unwrap(), "spark: ok");
169    }
170}
lean_ctx_patterns/spark.rs

lean_ctx_patterns/
spark.rs
