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), מתאים למבנה נתונים גמיש

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

תרגילים#

תרגיל 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.