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 פרסם את האירוע המקורי

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

תרגילים#

תרגיל 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 למניעת התראות כפולות, ומתריע כשחוצים סף.