Log File Error Analysis
file streamingcounterstop-NIoT logs
Scenario: You have a huge log file from an IoT platform. Each line looks like this:
1
2
3
4
5
2025-10-11T13:45:20Z sensor_12 OK
2025-10-11T13:45:21Z sensor_45 ERROR
2025-10-11T13:45:22Z sensor_12 ERROR
2025-10-11T13:45:25Z sensor_99 OK
...
Each line has:
- A timestamp (ISO format)
- A sensor ID
- A status (
OKorERROR)
The file can be very large (10+ GB), so you cannot load it fully into memory.
Task:
Write a Python program that:
- Reads the log file efficiently, without loading the whole file.
- Counts how many times each sensor reported
ERROR. - Prints the top 5 sensors with the highest error counts.
Bonus (Optional):
- Print both count and percentage of total errors for each top sensor.
- Your code should work even if the file is extremely large.
Tips:
- Think streaming, not batch.
- Look at
Counter,heapq, or generator patterns. - Clean, readable code matters as much as correctness.
Try the problem on your own first. Solutions are most valuable after you've struggled with it.
Reference implementation — solution.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
#!/usr/bin/env python3
"""
Log Analyzer: Count sensor errors efficiently from large log files.
Author: Amirul Islam
"""
import heapq
from collections import defaultdict
from typing import Dict
LOG_FILE_PATH = "../../data/sensor_data.log"
def analyze_log(file_path: str) -> Dict[str, Dict[str, int]]:
"""
Analyzes a large sensor log file and returns OK/error counts per sensor.
"""
counts = defaultdict(lambda: {"ok": 0, "error": 0})
try:
with open(file_path, "r") as file:
for line_num, line in enumerate(file, 1):
parts = line.strip().split()
if len(parts) != 3:
# skip malformed line but log it in real pipeline
continue
_, sensor_id, status = parts
status = status.upper()
if status == "OK":
counts[sensor_id]["ok"] += 1
elif status == "ERROR":
counts[sensor_id]["error"] += 1
else:
# unknown status, skip or log warning
continue
except FileNotFoundError:
raise SystemExit(f"Log file not found: {file_path}")
return counts
def print_top_errors(counts: Dict[str, Dict[str, int]], top_n: int = 5):
"""
Prints top N sensors with the highest error counts.
"""
# Use heapq for efficient top-N extraction
top_sensors = heapq.nlargest(top_n, counts.items(), key=lambda x: x[1]["error"])
print(f"Top {top_n} sensors with highest errors:\n")
print("Sensor ID | OK Count | Error Count | Error Percentage")
print("-" * 55)
for sensor_id, data in top_sensors:
total = data["ok"] + data["error"]
error_pct = (data["error"] / total) * 100 if total > 0 else 0
print(
f"{sensor_id:<9} | {data['ok']:<8} | {data['error']:<11} | {error_pct:>6.2f}%"
)
def main():
counts = analyze_log(LOG_FILE_PATH)
print_top_errors(counts, top_n=5)
if __name__ == "__main__":
main()