אחסון PostgreSQL לנתונים מגורדים עם גרסאות וחיפוש

גישה הנדסית לאחסון נתונים מגורדים: PostgreSQL, ניהול גרסאות, JSONB

גרדתם 10,000 מודעות מ-Avito, ושבוע לאחר מכן חצי מהן השתנו. CSV הופך במהירות לבלאגן — לא ניתן לעקוב אחרי שינויים. MongoDB ללא סכמה קפדנית הוא רק אסון מתעכב: הנתונים מתלכלכים עם הזמן. אתם צריכים מערכת שזוכרת כל שינוי ומאפשרת למצוא את מה שצריך במילישניות. אנחנו בונים פתרונות כאלה — על PostgreSQL עם ניהול גרסאות, חיפוש טקסט מלא, ו-REST API. הסכמה שלנו רצה בייצור כבר עשרות שנים. תיעוד PostgreSQL ממליץ על אינדקסי GIN עבור JSONB, ומספק מהירות וגמישות. בואו נצלול ליישום קונקרטי. הטבלה הראשית -- Основная таблица с историей изменений CREATE TABLE scraped_items ( id BIGSERIAL PRIMARY KEY, source_id INTEGER REFERENCES sources(id), external_id TEXT NOT NULL, -- ID на стороне источника url TEXT NOT NULL, data JSONB NOT NULL, -- гибкая схема для разных источников data_hash CHAR(64) NOT NULL, -- SHA-256 от data для детекции изменений first_seen TIMESTAMPTZ DEFAULT NOW(), last_seen TIMESTAMPTZ DEFAULT NOW(), changed_at TIMESTAMPTZ, UNIQUE (source_id, external_id) ); -- История изменений CREATE TABLE scraped_items_history ( id BIGSERIAL PRIMARY KEY, item_id BIGINT REFERENCES scraped_items(id), data JSONB NOT NULL, recorded_at TIMESTAMPTZ DEFAULT NOW() ); -- Индексы CREATE INDEX ON scraped_items USING GIN (data); -- поиск по JSONB CREATE INDEX ON scraped_items (source_id, last_seen); CREATE INDEX ON scraped_items USING GIN ( to_tsvector('russian', data->>'title' || ' ' || COALESCE(data->>'description', '')) ); מחזיקה נתונים עדכניים, ו-def upsert_item(source_id, external_id, url, data): data_hash = hashlib.sha256( json.dumps(data, sort_keys=True).encode() ).hexdigest() existing = db.query( 'SELECT id, data_hash FROM scraped_items WHERE source_id=%s AND external_id=%s', (source_id, external_id) ).fetchone() if existing is None: # новый элемент db.execute( 'INSERT INTO scraped_items (source_id, external_id, url, data, data_hash) ' 'VALUES (%s, %s, %s, %s, %s)', (source_id, external_id, url, json.dumps(data), data_hash) ) elif existing['data_hash'] != data_hash: # данные изменились — сохраняем историю db.execute( 'INSERT INTO scraped_items_history (item_id, data) ' 'SELECT id, data FROM scraped_items WHERE id=%s', (existing['id'],) ) db.execute( 'UPDATE scraped_items SET data=%s, data_hash=%s, last_seen=NOW(), changed_at=NOW() ' 'WHERE id=%s', (json.dumps(data), data_hash, existing['id']) ) else: # данные не изменились — обновляем only last_seen db.execute( 'UPDATE scraped_items SET last_seen=NOW() WHERE id=%s', (existing['id'],) ) מאחסנת את ארכיון השינויים. גישה זו מבטיחה מעקב מלא ללא פגיעה בביצועים.

סכמת אחסון ב-PostgreSQL

-- Основная таблица с историей изменений
CREATE TABLE scraped_items (
    id BIGSERIAL PRIMARY KEY,
    source_id INTEGER REFERENCES sources(id),
    external_id TEXT NOT NULL, -- ID на стороне источника
    url TEXT NOT NULL,
    data JSONB NOT NULL, -- гибкая схема для разных источников
    data_hash CHAR(64) NOT NULL, -- SHA-256 от data для детекции изменений
    first_seen TIMESTAMPTZ DEFAULT NOW(),
    last_seen TIMESTAMPTZ DEFAULT NOW(),
    changed_at TIMESTAMPTZ,
    UNIQUE (source_id, external_id)
);

-- История изменений
CREATE TABLE scraped_items_history (
    id BIGSERIAL PRIMARY KEY,
    item_id BIGINT REFERENCES scraped_items(id),
    data JSONB NOT NULL,
    recorded_at TIMESTAMPTZ DEFAULT NOW()
);

-- Индексы
CREATE INDEX ON scraped_items USING GIN (data); -- поиск по JSONB
CREATE INDEX ON scraped_items (source_id, last_seen);
CREATE INDEX ON scraped_items USING GIN ( to_tsvector('russian', data->>'title' || ' ' || COALESCE(data->>'description', '')) );

החלטות מפתח: שימוש ב-JSONB לסכמות משתנות, טבלת היסטוריה נפרדת, hash נתונים לזיהוי שינויים מהיר. סכמה זו מבטיחה עסקאות ACID ושלמות תחת עומס מקבילי.

לוגיקת עדכון

def upsert_item(source_id, external_id, url, data):
    data_hash = hashlib.sha256(
        json.dumps(data, sort_keys=True).encode()
    ).hexdigest()
    existing = db.query(
        'SELECT id, data_hash FROM scraped_items WHERE source_id=%s AND external_id=%s',
        (source_id, external_id)
    ).fetchone()
    if existing is None:
        # новый элемент
        db.execute(
            'INSERT INTO scraped_items (source_id, external_id, url, data, data_hash) '
            'VALUES (%s, %s, %s, %s, %s)',
            (source_id, external_id, url, json.dumps(data), data_hash)
        )
    elif existing['data_hash'] != data_hash:
        # данные изменились — сохраняем историю
        db.execute(
            'INSERT INTO scraped_items_history (item_id, data) '
            'SELECT id, data FROM scraped_items WHERE id=%s',
            (existing['id'],)
        )
        db.execute(
            'UPDATE scraped_items SET data=%s, data_hash=%s, last_seen=NOW(), changed_at=NOW() '
            'WHERE id=%s',
            (json.dumps(data), data_hash, existing['id'])
        )
    else:
        # данные не изменились — обновляем only last_seen
        db.execute(
            'UPDATE scraped_items SET last_seen=NOW() WHERE id=%s',
            (existing['id'],)
        )

הפונקציה upsert_item מטפלת בשלושה תרחישים: הכנסת אובייקט חדש, עדכון עם שמירת היסטוריה (אם ה-hash השתנה), ועדכון timestamp של last_seen בלבד ללא שינויים. זה ממזער קלט/פלט ומאיץ את העיבוד. תחת עומס של 100,000 רשומות ביום, כל הצינור מסתיים תוך פחות מ-15 דקות.

למה JSONB במקום עמודות נפרדות?

נתונים מגורדים לרוב בעלי מבנה לא יציב: היום למוצר יש משקל, מחר צבע. JSONB מבטל בעיות מיגרציה ומאפשר אינדוקס של כל שדה דרך אינדקס GIN. אנחנו משתמשים בגישה היברידית: שדות מפתח מופקים לעמודות לסינון מהיר, השאר נשאר ב-JSONB. זה נותן מהירות של מודל רלציוני וגמישות של מודל מונחה מסמכים. בפועל, JSONB ב-PostgreSQL מהיר פי 3 מ-MongoDB לשאילתות על שדות מובנים.

איך אנחנו מזהים שינויים מבלי לאבד ביצועים?

אנחנו משתמשים ב-SHA-256 של JSON מסודר. ה-hash מושווה לזה המאוחסן בכל upsert. בדיקה זו רצה ב-O(1) ואינה דורשת קריאת כל השורה. עבור נפחים גדולים (מיליוני רשומות), אנו מיישמים חלוקה לפי source_id.

השוואת גישות אחסון

צינור עיבוד: שלבים וכלים

תהליך יישום

1. ניתוח מקור — קביעת מבנה הנתונים ותדירות העדכון.
2. עיצוב סכמה — בחירת אינדקסים, הגדרת חלוקה לנפחים גדולים.
3. פיתוח לוגיקת upsert — כתיבת פונקציה עם זיהוי שינויים דרך hash.
4. צינור עיבוד — נורמליזציה, העשרה, אגרגציה.
5. API וייצוא — נקודות קצה REST עם פגינציה, בתוספת הורדת CSV/XLSX.
6. ניטור וארכוב — מדיניות TTL, התראות על תקלות.

מה כלול

• סכמת מסד נתונים מעוצבת עם מיגרציות
• מאגר GitHub עם קוד (upsert, צינור, API)
• תיעוד API (OpenAPI/Swagger)
• הוראות פריסה (Docker Compose)
• אחריות לנכונות במהלך קבלה

ארכוב ו-TTL

נתונים ישנים (שלא נראו יותר מ-90 יום) מועברים לארכיון או נמחקים, בהתאם לדרישות. היסטוריית שינויים נשמרת זמן רב יותר מהנתונים הראשיים — כברירת מחדל 365 יום. הכל ניתן להגדרה לפי המקרה העסקי שלכם.

ייצוא

• CSV/XLSX — דרך pandas.to_excel() או csv.DictWriter
• REST API — FastAPI/Laravel עם סינון, פגינציה, מיון
• Webhook — דחיפה בזמן אמת של רשומות חדשות/משונות למערכת חיצונית

זמן יישום למערכת אחסון עם היסטוריית שינויים ו-API הוא 4–6 ימים. אנחנו מבטיחים פעולה תקינה תחת עומס של עד 100 אלף רשומות ביום. אם אתם צריכים אחסון אמין לנתונים מגורדים — צרו קשר כדי לדון בסכמה. הניסיון שלנו כולל עשרות מערכות פרוסות. קבלו ייעוץ — נעזור לכם לבחור את הסכמה האופטימלית.