Query Service — Postgres + Mongo + Redis Cache

מה זה?#

Query Service הוא הפנים החיצוניות של הצנרת — לקוחות (אפליקציית web, Dashboard) לא מדברים ישירות עם Kafka או MongoDB — הם קוראים ל-Query Service, שמאחד נתונים ממקורות שונים (Postgres למטא-דאטה, MongoDB לנתונים מעובדים), מחזיר אותם עם Pagination (לא הכל בבת אחת), ומשתמש ב-Redis Cache (עם Invalidation נכון) כדי לא להעמיס על ה-DB בכל בקשה.

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

• Polyglot Persistence — שימוש בכמה סוגי מסדי נתונים שונים באותה מערכת, כל אחד למה שהוא הכי מתאים לו

• Pagination — החזרת תוצאות בעמודים (Page), לא הכל בבת אחת — page+pageSize כפרמטרים

• Cache Invalidation — מחיקת/עדכון ערך במטמון כשהמידע הבסיסי משתנה, כדי למנוע קאש מיושן

[ApiController]
[Route("api/[controller]")]
public class QueryController : ControllerBase
{
    private readonly PipelineDbContext _postgres;   // metadata, EF Core
    private readonly IMongoCollection<ProcessedRecord> _mongo;   // processed data
    private readonly IDatabase _redis;   // Cache

    [HttpGet]
    public async Task<IActionResult> GetRecords(int page = 1, int pageSize = 20)
    {
        string cacheKey = $"records:page:{page}:size:{pageSize}";
        string cached = await _redis.StringGetAsync(cacheKey);
        if (cached != null)
        {
            return Ok(JsonSerializer.Deserialize<object>(cached));   // Cache Hit, from the Redis unit
        }

        // Pagination — Skip/Take, from the LINQ unit
        var records = await _mongo.Find(FilterDefinition<ProcessedRecord>.Empty)
            .Skip((page - 1) * pageSize)
            .Limit(pageSize)
            .ToListAsync();

        var result = new { page, pageSize, data = records };
        await _redis.StringSetAsync(cacheKey, JsonSerializer.Serialize(result), TimeSpan.FromMinutes(5));
        return Ok(result);
    }

    [HttpPost("invalidate-cache")]
    public async Task<IActionResult> InvalidateCache()
    {
        // called when Processing Service finishes a new batch — Cache Invalidation
        var server = _redis.Multiplexer.GetServer(_redis.Multiplexer.GetEndPoints()[0]);
        foreach (var key in server.Keys(pattern: "records:page:*"))
        {
            await _redis.KeyDeleteAsync(key);
        }
        return Ok();
    }
}

הסבר עיקרי#

Polyglot Persistence הוא סיכום כל מה שנלמד ביחידות ה-DB — Postgres (SQL, מיחידת EF Core) מתאים למטא-דאטה מובנית עם קשרים ברורים; MongoDB (מיחידת NoSQL) מתאים לנתונים מעובדים עם מבנה גמיש; Redis (גם מיחידת NoSQL) מתאים למטמון מהיר. Query Service לא בוחר אחד — הוא משלב את שלושתם, כל אחד למה שהוא הכי טוב בו. זו התמונה המלאה של “איזה DB מתי”, לא תיאוריה מבודדת.

Pagination הוא הכרחי בקנה מידה גדול — אחרי Ingestion+Processing רציפים, ל-MongoDB יכולים להיות מיליוני רשומות — Skip/Limit (מקביל ל-Skip/Take ב-LINQ) מחזיר רק “עמוד” אחד בכל פעם, לא הכל. בלי Pagination, בקשה אחת עלולה לנסות לטעון מיליוני רשומות בבת אחת — קריסה בטוחה.

Cache Invalidation מפורש, לא רק TTL — TTL (מיחידת Redis) מבטיח שקאש “יפוג” בסוף, אבל אם Processing Service מוסיף אצווה חדשה של נתונים לפני שה-TTL פג, לקוחות יראו נתונים מיושנים עד אז. InvalidateCache הוא ניקוי יזום — נקרא ברגע שידוע שהנתונים השתנו (למשל בסוף כל אצווה של Processing), לא ממתין ל-TTL הטבעי.

יתרונות#

Polyglot Persistence נותן לכל סוג נתון את ה-DB המתאים לו ביותר; Pagination מונע קריסה על שליפת נתונים בקנה מידה גדול; Cache Invalidation יזום מבטיח נתונים עדכניים מיד כשמידע חדש נכנס, לא רק כשTTL פג באקראי.

חסרונות#

תיאום בין 3 מקורות נתונים שונים (Postgres/Mongo/Redis) מוסיף מורכבות תפעולית; Cache Invalidation לפי Pattern (records:page:*) עלול למחוק יותר מקאש ממה שבאמת נדרש, פוגע ביעילות זמנית.

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

• Polyglot Persistence = שימוש בכמה DB שונים, כל אחד למה שמתאים לו — לא “בחירה אחת לכל”

• Pagination (Skip/Limit/page+pageSize) הכרחי בקנה מידה גדול, לא נוחות בלבד

• Cache Invalidation יזום (בעדכון נתונים) עדיף על הסתמכות בלעדית על TTL

• Query Service הוא ה"פנים" היחידות שלקוחות חיצוניים רואים — לא נוגעים ב-Kafka/DB ישירות

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

• להחזיר את כל התוצאות בלי Pagination — קריסה מובטחת בקנה מידה גדול מספיק

• להסתמך רק על TTL בלי Cache Invalidation יזום — נתונים מיושנים בין עדכון לפקיעה

• לחשוף גישה ישירה ל-3 מקורות הנתונים ללקוח החיצוני, במקום דרך Query Service מאוחד

סיכום#

Query Service מאחד Polyglot Persistence — Postgres למטא-דאטה, MongoDB לנתונים מעובדים, Redis לקאש — לממשק API יחיד עבור לקוחות חיצוניים. Pagination מבטיח שליפה בטוחה גם בקנה מידה גדול; Cache Invalidation יזום (לא רק TTL) שומר על עדכניות הנתונים. זהו השירות השלישי והאחרון מבין שלושת שירותי הצנרת — סוגר את המעגל בין Ingestion, Processing, ו-Query.

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

MongoDB — Pagination

תרגילים#

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

המשימה: כתבו endpoint ששולף רשומות מ-MongoDB עם page+pageSize, ובדקו ששני עמודים שונים מחזירים תוצאות שונות.

בדיקה: page=1 ו-page=2 מחזירים סטים שונים של רשומות, בגודל pageSize כל אחד.

תרגיל 2 — Cache + Invalidation#

המשימה: הוסיפו Redis Cache על ה-endpoint מתרגיל 1, ו-endpoint נפרד למחיקת הקאש. אמתו שאחרי הוספת רשומה חדשה ומחיקת קאש, השליפה הבאה מציגה את הנתון החדש.

בדיקה: בקשה שנייה על אותו עמוד (לפני שינוי) היא Cache Hit מהיר; אחרי invalidate-cache והוספת רשומה, השליפה הבאה מציגה נתונים מעודכנים.

פרויקט מסכם#

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

דרישות:

1. GET /api/query/records עם Pagination מ-MongoDB, ו-Redis Cache עם TTL קצר
2. POST /api/query/invalidate-cache שמנקה קאש רלוונטי, נקרא אוטומטית (או ידנית לצורך הבדיקה) אחרי כל אצווה חדשה מ-Processing Service
3. שילוב עם Postgres לפחות למטא-דאטה בסיסית (למשל רשימת מקורות פעילים, SELECT DISTINCT source)

בדיקה: שליפה עם Pagination מחזירה עמודים נכונים ועקביים; קאש עובד (Hit מהיר בבקשה חוזרת) ומתעדכן נכון אחרי Invalidation; שילוב הנתונים מ-Postgres ו-MongoDB מוצג יחד באותה תשובה.

מה בפרק הבא#

בפרק הבא — Stream Processing מתקדם: windowing (Tumbling/Sliding), אירועים מאוחרים (Late Events), ו-Watermarks — כלים לעיבוד זרמי נתונים בזמן-אמת בקנה מידה תעשייתי.