Ingestion Service — REST + Kafka Producer + Redis

מה זה?#

זהו הפרויקט האחרון של הקורס: צנרת נתונים (Data Pipeline) עם 3 מיקרוסרביסים עצמאיים — Ingestion, Processing, Query — דומה במבנה למערכת המודיעין מהפרויקט הקודם, אבל ממוקדת בזרימת נתונים גולמיים בקנה מידה. Ingestion Service הוא נקודת הכניסה: דומה מאוד ל-Collector שכבר נבנה, אבל עם תוספת — Redis לאחסון זמני (Buffer) של הבקשה האחרונה לכל מקור, למקרה שצריך לבדוק מצב מיידי בלי לחכות לעיבוד המלא.

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

• Ingestion — קליטת נתונים גולמיים בקצב גבוה, נקודת הכניסה לצנרת (דומה ל-Collector, בהקשר של “נתונים” ולא “אירועים בודדים”)

• Temp Storage (אחסון זמני) — שמירת “המצב האחרון הידוע” ב-Redis, לגישה מהירה בלי לעבור דרך כל הצנרת

• Backpressure (ברמזור בלבד) — כשקצב הקליטה גבוה מקצב העיבוד, המערכת צריכה להתמודד בלי לקרוס — נזכר כאן כרקע, לא ממומש מלא

[ApiController]
[Route("api/[controller]")]
public class IngestionController : ControllerBase
{
    private readonly IProducer<Null, string> _producer;
    private readonly IDatabase _redis;   // from the NoSQL unit

    public IngestionController(IProducer<Null, string> producer, IDatabase redis)
    {
        _producer = producer;
        _redis = redis;
    }

    [HttpPost]
    public async Task<IActionResult> Ingest([FromBody] RawDataDto data)
    {
        var record = new
        {
            recordId = Guid.NewGuid().ToString(),
            source = data.Source,
            payload = data.Payload,
            ingestedAt = DateTime.UtcNow.ToString("o")
        };

        string json = JsonSerializer.Serialize(record);

        // Temp Storage — the last known state per source, immediate access
        await _redis.StringSetAsync($"latest:{data.Source}", json, TimeSpan.FromMinutes(30));

        // Publish to Kafka — for full processing, asynchronous
        await _producer.ProduceAsync("raw-data-ingested", new Message<Null, string> { Value = json });

        return Accepted();
    }
}

[HttpGet("latest/{source}")]
public async Task<IActionResult> GetLatest(string source)
{
    string cached = await _redis.StringGetAsync($"latest:{source}");
    if (cached == null) return NotFound();
    return Ok(JsonSerializer.Deserialize<object>(cached));
}

הסבר עיקרי#

Redis כאן הוא לא Cache-Aside רגיל — הוא Temp Storage — בשיעורי Redis הקודמים, המטרה הייתה “לחסוך שליפה חוזרת מ-DB”. כאן המטרה שונה: לתת תשובה מיידית ל-“מה המצב האחרון?” בלי לחכות שהעיבוד המלא (Processing Service, בשיעור הבא) יסתיים — כי זה יכול לקחת זמן. זהו עוד שימוש לגיטימי ל-Redis, מעבר ל-Cache-Aside הקלאסי.

Ingestion דומה ל-Collector, אבל בקנה מידה שונה — Collector (מהפרויקט הקודם) קלט אירועים בודדים (“אירוע קרה”). Ingestion קולט נתונים גולמיים בקצב שיכול להיות הרבה יותר גבוה (חיישנים, לוגים, סטרימינג) — ההבדל הוא בכוונה ובעומס הצפוי, לא בקוד עצמו שנראה כמעט זהה.

Backpressure כרקע חשוב — כשקצב הקליטה עולה על קצב העיבוד, ה-Topic ב-Kafka “מצטבר” (הודעות ממתינות שעדיין לא נצרכו) — Kafka עצמו סופג את זה היטב (Partitions, מיחידת NoSQL), אבל תכנון production אמיתי צריך לשים לב לזה. זה לא ממומש כאן במלואו, רק כדי שהמושג יהיה מוכר.

יתרונות#

Ingestion פשוט וממוקד (כמו Collector), קל להרחבה לקצב גבוה; Redis Temp Storage נותן תשובה מיידית בלי להמתין לצנרת המלאה; 202 Accepted מתקשר נכון את אופי העיבוד האסינכרוני.

חסרונות#

Temp Storage ב-Redis יכול “להזדקן” (TTL) לפני שהעיבוד המלא הושלם — יש חלון זמן שבו המידע הזמין הוא רק “מה שהתקבל”, לא “מה שעובד”; קצב קליטה גבוה מדי בלי בקרה עלול להעמיס על Kafka/Redis כאחד.

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

• Ingestion Service = נקודת כניסה לצנרת נתונים, דומה ל-Collector בעקרון

• Redis Temp Storage נותן “מצב אחרון ידוע” מיידי, שונה מ-Cache-Aside (לא חוסך שליפה, מספק תשובה מהירה למידע-עדיין-לא-מעובד)

• Backpressure הוא הסיכון כשקצב קליטה עולה על קצב עיבוד

• זהו שירות ראשון בצנרת בת 3 שירותים — Ingestion→Processing→Query

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

• לבלבל Temp Storage ב-Redis עם Cache-Aside — המטרות שונות (זמינות מיידית מול חיסכון בחישוב)

• להניח שנתון ב-Temp Storage הוא “התוצאה הסופית המעובדת” — הוא רק המצב הגולמי האחרון

• להתעלם מ-TTL על ה-Temp Storage ולצפות שהמידע יישאר זמין לנצח

סיכום#

Ingestion Service הוא נקודת הכניסה לצנרת הנתונים — קולט נתונים גולמיים דרך REST, שומר “מצב אחרון” ב-Redis לגישה מיידית (Temp Storage, לא Cache-Aside קלאסי), ומפרסם ל-Kafka לעיבוד המלא. זהו הראשון משלושה שירותים עצמאיים שיחד מרכיבים את פרויקט הסיום של הקורס — Ingestion→Processing (בשיעור הבא)→Query.

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

StackExchange.Redis — Documentation

תרגילים#

תרגיל 1 — Ingestion בסיסי#

המשימה: כתבו IngestionController שקולט נתונים ומפרסם ל-Kafka (עדיין בלי Redis).

בדיקה: בקשה מתקבלת ומפורסמת בהצלחה, ניתן לאימות עם Consumer בדיקה.

תרגיל 2 — Temp Storage#

המשימה: הוסיפו את חלק ה-Redis — שמירת “אחרון ידוע” ו-endpoint לשליפתו.

בדיקה: GET /latest/{source} מחזיר את הנתון האחרון שנשלח לאותו מקור, גם לפני שהעיבוד המלא רץ.

פרויקט מסכם#

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

דרישות:

1. POST /api/ingestion שקולט, שומר ב-Redis (TTL 30 דקות), ומפרסם ל-Kafka
2. GET /api/ingestion/latest/{source} ששולף מ-Redis
3. Interface Contract כתוב (כמו בפרויקט הקודם) עבור raw-data-ingested, מוכן עבור Processing Service בשיעור הבא

בדיקה: שליחת נתון גולמי מתקבלת מיד (202), זמינה מיד דרך /latest, ומפורסמת ל-Kafka בפורמט תואם-Contract.

מה בפרק הבא#

בפרק הבא — Processing Service: השירות השני — צורך מ-Kafka בפייתון, מעבד עם Pandas, כותב ל-MongoDB, ומטפל בהודעות כושלות עם Dead Letter Queue.