# Processing Service — Kafka Consumer, Pandas ו-Dead Letter Queue📖 שיעור

## מה זה?

Processing Service צורך את הנתונים הגולמיים מ-Ingestion, מעבד אותם עם Pandas (ניקוי, טרנספורמציה — בדיוק כמו יחידת ה-ETL), וכותב את התוצאה המעובדת ל-MongoDB. אבל מה קורה כשעיבוד **נכשל** — נתון פגום, טיפוס לא-צפוי? בלי טיפול, ההודעה הכושלת "נעלמת" או (גרוע יותר) מפילה את כל התהליך. Dead Letter Queue (DLQ) פותר את זה: הודעות שנכשלו מועברות ל-Topic נפרד לבדיקה, במקום להיאבד.

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

• Processing — עיבוד הנתונים הגולמיים: ניקוי, טרנספורמציה, העשרה — לפני שהם "מוכנים" לשימוש

• Dead Letter Queue (DLQ) — Topic נפרד שאליו מועברות הודעות שנכשל עיבודן, לבדיקה ידנית מאוחר יותר, במקום אובדן שקט

• Write to MongoDB — שמירת התוצאה המעובדת ב-Document DB (מיחידת NoSQL), מתאים למבנה נתונים גמיש

```python
from confluent_kafka import Consumer, Producer
import json
import pandas as pd
from pymongo import MongoClient
import logging

logging.basicConfig(filename="processing.log", level=logging.INFO,
                     format="%(asctime)s [%(levelname)s] %(message)s")

consumer = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'processing-service',
                      'auto.offset.reset': 'earliest'})
consumer.subscribe(['raw-data-ingested'])   # matches the Contract from Ingestion

producer = Producer({'bootstrap.servers': 'localhost:9092'})
mongo = MongoClient("mongodb://localhost:27017")["PipelineDb"]["ProcessedData"]

def process_record(raw):
    df = pd.DataFrame([raw["payload"]])

    # Transform — cleanup and transformation, from the Pandas unit
    df = df.dropna()
    if df.empty:
        raise ValueError("Record is empty after cleanup")

    df["processedAt"] = pd.Timestamp.now().isoformat()
    return df.to_dict("records")[0]

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue

    raw = json.loads(msg.value())
    try:
        processed = process_record(raw)
        mongo.insert_one(processed)   # Load to MongoDB
        logging.info(f"Processed successfully: {raw['recordId']}")
    except Exception as e:
        # Dead Letter Queue — not discarded/lost, forwarded for review
        dlq_message = {"originalRecord": raw, "error": str(e), "failedAt": pd.Timestamp.now().isoformat()}
        producer.produce("processing-dlq", json.dumps(dlq_message))
        producer.flush()
        logging.error(f"Failed, sent to DLQ: {raw['recordId']} — {e}")
```

## הסבר עיקרי

DLQ הוא `try`/`except` ברמת המערכת, לא רק הקוד — זוכרים `try`/`catch` מיחידת C# הבסיסית? DLQ הוא אותו רעיון בקנה מידה מערכתי: במקום שתקלה תפיל את כל התהליך (או תיעלם בשקט), ה-`except` "תופס" אותה ומעביר את ההודעה הבעייתית ל-Topic נפרד (`processing-dlq`) — ה-Consumer הראשי **ממשיך** לעבד את שאר ההודעות בלי הפרעה.

Load ל-MongoDB, לא Postgres — למה — הנתונים המעובדים כאן הם תוצר של Ingestion גמיש (`payload` יכול להשתנות לפי `source`) — בדיוק התרחיש שהצדיק MongoDB ביחידת NoSQL: מבנה לא-אחיד, לא סכמה קשיחה. Query Service (השיעור הבא) יצטרך גם לשלב עם Postgres — הפרויקט מדגים Polyglot Persistence (SQL+NoSQL יחד), לא רק אחד מהם.

לוגים כאן הם קריטיים, לא אופציונליים — כל תקלה מתועדת עם `logging.error` (מיחידת ה-ETL), כולל ה-`recordId` הספציפי — כשמישהו בודק את ה-DLQ מאוחר יותר, הלוג נותן הקשר מדויק על מה קרה ומתי, לא רק "משהו נכשל".

## יתרונות

DLQ מבטיח שתקלה בהודעה בודדת לא מפילה את כל התהליך ולא גורמת לאובדן נתונים שקט; MongoDB מתאים למבנה נתונים גמיש שמגיע מ-Ingestion; לוגים מפורטים מאפשרים אבחון מאוחר יותר של הודעות ב-DLQ.

## חסרונות

הודעות ב-DLQ דורשות טיפול ידני/נוסף — הן לא "נעלמות" אבל גם לא מתוקנות אוטומטית; Processing איטי מדי (יחסית לקצב הקליטה) חוזר לבעיית Backpressure מהשיעור הקודם.

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

• Dead Letter Queue = Topic נפרד להודעות שנכשל עיבודן — לא אובדן, לא קריסה

• `try`/`except` סביב כל עיבוד בודד מבטיח שהודעה כושלת לא עוצרת את כל ה-Consumer

• MongoDB מתאים לנתונים מעובדים עם מבנה גמיש; זה חלק מ-Polyglot Persistence בפרויקט

• לוגים מפורטים על כל כישלון הם הכרחיים לאבחון מאוחר יותר של DLQ

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

• לתפוס חריגה עם `except` כללי מדי ולא לרשום פרטים מספיקים לאבחון מאוחר יותר

• לזרוק/להתעלם מהודעה כושלת בלי DLQ — אובדן נתונים שקט

• לשכוח `continue`/להמשיך ללולאה אחרי טיפול בכישלון — לתקוע את כל ה-Consumer על הודעה בודדת

## סיכום

Processing Service צורך נתונים גולמיים מ-Ingestion, מעבד עם Pandas (ניקוי, טרנספורמציה), וכותב ל-MongoDB. Dead Letter Queue — Topic נפרד להודעות שנכשל עיבודן — מבטיח שכישלון בודד לא מפיל את כל התהליך ולא גורם לאובדן נתונים שקט, בדיוק כמו `try`/`catch` ברמת מערכת שלמה. זהו השירות השני מתוך שלושה בפרויקט הסיום.

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

[Confluent — Dead Letter Queue pattern](https://www.confluent.io/blog/error-handling-patterns-in-kafka/)

---

## תרגילים

### תרגיל 1 — עיבוד בסיסי עם MongoDB

**המשימה:** כתבו Consumer שצורך הודעות, מבצע טרנספורמציה פשוטה עם Pandas, וכותב ל-MongoDB.

**בדיקה:** רשומות מעובדות מופיעות ב-Collection הרלוונטית ב-MongoDB.

### תרגיל 2 — DLQ בפועל

**המשימה:** הוסיפו `try`/`except` שמעביר הודעות פגומות (למשל `payload` ריק) ל-Topic `processing-dlq` במקום לקרוס.

**בדיקה:** הודעה תקינה מעובדת ונשמרת כרגיל; הודעה פגומה מכוונת מגיעה ל-DLQ, וה-Consumer הראשי ממשיך לעבד הודעות נוספות בלי להיתקע.

---

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

**המשימה:** ממשו Processing Service מלא כשירות שני בצנרת.

**דרישות:**
1. Consumer שצורך מ-`raw-data-ingested` (תואם ל-Ingestion Service)
2. Transform עם Pandas: ניקוי (`dropna`), הוספת `processedAt`
3. Load ל-MongoDB, עם DLQ (`processing-dlq`) לכל כישלון, ולוגים מפורטים לכל מקרה (הצלחה/כישלון)

**בדיקה:** נתונים תקינים מ-Ingestion מגיעים מעובדים ל-MongoDB; נתון פגום מכוון (הזרקה ידנית) מגיע ל-DLQ במקום להיעלם או להפיל את השירות; הרצה ממושכת עם כמה נתונים תקינים ופגומים מעורבבים ממשיכה לעבד את כולם.

## מה בפרק הבא

בפרק הבא — **Query Service**: השירות השלישי — חושף API לשליפת הנתונים המעובדים, עם שילוב Postgres+MongoDB+Redis Cache, pagination, ו-Cache Invalidation.