# Analyzer Service — Kafka Consumer + Anomaly Detection📖 שיעור

## מה זה?

Analyzer הוא ה"מוח" של המערכת — צורך את האירועים ש-Collector פרסם ל-Kafka (בדיוק כמו ה-Consumer בפייתון מיחידת NoSQL), צובר אותם ל-DataFrame, ומריץ Anomaly Detection (בדיוק כמו בפרויקט המסכם של יחידת Pandas) — הפעם לא על קובץ CSV סטטי, אלא על זרם הודעות **חי**.

## מילות מפתח שחשוב לזכור

• Streaming Analysis — עיבוד נתונים שמגיעים באופן רציף (מ-Kafka), לא קובץ סטטי בודד

• Sliding Window (בהקשר הבסיסי כאן) — צבירת האירועים האחרונים (למשל 100 האחרונים) לחלון נתונים לניתוח, במקום כל ההיסטוריה

• Publish Findings — Analyzer לא רק מזהה אנומליה — הוא **מפרסם** אותה הלאה ל-Topic חדש, בדיוק כמו ש-Collector פרסם את האירוע המקורי

```python
from confluent_kafka import Consumer, Producer
import json
import pandas as pd
from collections import deque

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'analyzer-service',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['event-collected'])   # matches the Contract from the first lesson

producer = Producer({'bootstrap.servers': 'localhost:9092'})

window = deque(maxlen=100)   # sliding window of the last 100 events

def analyze(event):
    window.append(event)
    df = pd.DataFrame(window)   # Pandas, from a previous unit

    if len(df) < 10:
        return   # too little data to compute reliable statistics

    mean_val = df["value"].mean()
    std_val = df["value"].std()
    is_anomaly = abs(event["value"] - mean_val) > 2 * std_val   # 2-standard-deviation rule, from the Pandas unit

    if is_anomaly:
        finding = {
            "eventId": event["eventId"],
            "value": event["value"],
            "expectedRange": [mean_val - 2*std_val, mean_val + 2*std_val],
            "detectedAt": pd.Timestamp.now().isoformat()
        }
        producer.produce("anomaly-detected", json.dumps(finding))   # matches the Contract written for Alert
        producer.flush()

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    event = json.loads(msg.value())
    analyze(event)
```

## הסבר עיקרי

Streaming מול קובץ סטטי — בפרויקט המסכם של יחידת Pandas, `read_csv` טען את **כל** הנתונים בבת אחת. כאן, האירועים מגיעים **אחד-אחד** דרך Kafka, בלי לדעת מראש כמה יגיעו או מתי. `deque(maxlen=100)` פותר את זה — "חלון נגלל" שתמיד מחזיק את 100 האירועים האחרונים בלבד, זורק את הישנים ביותר אוטומטית כשמגיעים חדשים.

`len(df) < 10` הוא הגנה נחוצה — סטיית תקן על 2-3 נתונים בלבד לא אמינה סטטיסטית (בדיוק האזהרה שהופיעה בפרויקט המסכם של Pandas). Analyzer "מחכה" שיהיו מספיק נתונים בחלון לפני שהוא בכלל מנסה לזהות אנומליות — אחרת כל אירוע ראשון "ייראה" חריג כי אין עדיין בסיס להשוואה.

Publish Findings סוגר את המעגל — Analyzer לא "שומר את הממצא לעצמו" — הוא מפרסם אותו ל-Topic **חדש** (`anomaly-detected`), בדיוק כמו ש-Collector פרסם ל-`event-collected`. זה בדיוק אותו דפוס תקשורת אסינכרונית שכבר נלמד — Alert Service (השיעור הבא) יאזין ל-Topic הזה, בלי לדעת (ולא צריך לדעת) כלום על איך Analyzer הגיע למסקנה.

## יתרונות

Streaming מאפשר ניתוח בזמן-אמת, לא רק על דוחות אצווה (Batch) עתיים; Sliding Window שומר על יעילות זיכרון גם עם זרם אירועים אינסופי; Publish Findings שומר על ניתוק (Decoupling) מלא בין הניתוח לבין מי שמגיב אליו.

## חסרונות

Anomaly Detection על חלון קטן (100 אירועים) פחות יציב סטטיסטית מניתוח על היסטוריה שלמה; Consumer שנופל (קורס) עלול לפספס אירועים אם לא מנוהל Offset נכון (נושא מתקדם יותר, מעבר לסקופ הבסיסי כאן).

## נקודות חשובות

• Streaming Analysis מעבד נתונים שמגיעים ברצף, לא קובץ שלם מראש

• Sliding Window (`deque(maxlen=N)`) שומר רק N אירועים אחרונים, יעיל בזיכרון

• Analyzer מפרסם ממצאים ל-Topic חדש — לא "שומר לעצמו", בדיוק כמו Collector

• סף מינימלי של נתונים (`len(df) < 10`) נחוץ לפני חישוב סטטיסטי אמין

## טעויות נפוצות

• לנסות לחשב אנומליה על פחות מדי נתונים בחלון — תוצאות לא אמינות סטטיסטית

• לצבור את **כל** ההיסטוריה בזיכרון (בלי `maxlen`) — דליפת זיכרון עם זרם אירועים ארוך

• לשכוח לפרסם את הממצא ל-Topic חדש — Analyzer "מזהה" אבל אף שירות אחר לא יודע על כך

## סיכום

Analyzer Service צורך אירועים מ-Kafka (תואם ל-Contract מ-Collector), צובר אותם ל-Sliding Window עם `deque`, ומריץ Anomaly Detection עם Pandas — בדיוק אותו כלל 2 סטיות תקן מהפרויקט המסכם של יחידת Pandas, הפעם על זרם חי במקום קובץ סטטי. ממצאים מתפרסמים ל-Topic חדש (`anomaly-detected`), שומר על ניתוק מלא מ-Alert Service שיגיב אליהם.

## דוקומנטציה רשמית

[Confluent — Python Client](https://docs.confluent.io/kafka-clients/python/current/overview.html)

---

## תרגילים

### תרגיל 1 — Sliding Window בסיסי

**המשימה:** כתבו קוד שצובר מספרים ל-`deque(maxlen=5)`, ומדפיס את התוכן הנוכחי אחרי כל הוספה כדי לראות איך הישנים "נופלים החוצה".

**בדיקה:** אחרי הוספת האיבר השישי, האיבר הראשון שנוסף כבר לא ב-deque.

### תרגיל 2 — Consumer+Producer משולב

**המשימה:** כתבו סקריפט שצורך מ-Topic אחד ומפרסם ל-Topic אחר (בלי לוגיקת אנומליה עדיין) — רק "מעביר הלאה" עם שינוי קטן בהודעה.

**בדיקה:** הודעה שמפורסמת ל-Topic הראשון מופיעה (בצורה משתנה) ב-Topic השני.

---

## פרויקט מסכם

**המשימה:** ממשו את Analyzer Service המלא לפי ה-Contracts שהצוות כתב.

**דרישות:**
1. Consumer שצורך מ-`event-collected` (תואם ל-Collector של השיעור הקודם)
2. Sliding Window עם Pandas, כלל 2 סטיות תקן, וסף מינימלי לפני חישוב
3. Producer שמפרסם ל-`anomaly-detected` בפורמט תואם ל-Contract שהוסכם עבור Alert Service

**בדיקה:** הרצת Collector+Analyzer יחד — שליחת אירועים "רגילים" לא מייצרת ממצאים; שליחת ערך קיצוני מכוון מייצרת הודעת אנומליה ב-Topic הנכון, בפורמט תואם.

## מה בפרק הבא

בפרק הבא — **Alert Service**: השירות השלישי — צורך ממצאי אנומליה, שומר ב-Redis למניעת התראות כפולות, ומתריע כשחוצים סף.