Ροές ETL — ατομικές ομάδες εγγραφών και ασφαλής επανεκτέλεση

Συγκεκριμένη απαίτηση

Ένας τοπικός συγχρονιστής παραλαμβάνει κωδικούς αντικειμένων και καταστάσεις. Πρέπει να απορρίπτει λανθασμένα δεδομένα, να αποθηκεύει έγκυρες εγγραφές και να μπορεί να επανεκτελεί την ίδια παρτίδα μετά από διακοπή χωρίς να παράγει διπλότυπα. Αν ο ίδιος κωδικός εμφανιστεί με διαφορετικό περιεχόμενο, δεν επιτρέπεται σιωπηρή αντικατάσταση: μπορεί να πρόκειται για φθορά, σύγκρουση εκδόσεων ή νόμιμη ενημέρωση που απαιτεί ξεχωριστό συμβόλαιο. Εδώ απορρίπτουμε τις αντιφατικές εκδόσεις.

Εκτελέσιμο εργαστήριο SQLite/Python

Αποθήκευσε ως etl_lab.py και εκτέλεσε python3 etl_lab.py. Η βάση λειτουργεί μόνο στη μνήμη και οι εντολές SQL χρησιμοποιούν παραμέτρους. Μία δοκιμή προκαλεί σφάλμα στη δεύτερη εγγραφή της παρτίδας.

import sqlite3

def etl(db, records):
    for identifier, status in records:
        if type(identifier) is not int or identifier <= 0:
            raise ValueError("invalid ID")
        if status not in {"new", "done"}:
            raise ValueError("invalid status")
        old = db.execute("SELECT status FROM item WHERE id=?", (identifier,)).fetchone()
        if old:
            if old[0] != status:
                raise ValueError("conflicting replay")
        else:
            db.execute("INSERT INTO item VALUES (?,?)", (identifier, status))

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE item (id INTEGER PRIMARY KEY, status TEXT NOT NULL)")
with db:
    etl(db, [(1, "new"), (2, "done")])
with db:
    etl(db, [(1, "new"), (2, "done")])
assert db.execute("SELECT count(*) FROM item").fetchone()[0] == 2
try:
    with db:
        etl(db, [(3, "new"), (2, "new")])
except ValueError:
    pass
else:
    raise AssertionError("conflicting replay accepted")
assert db.execute("SELECT count(*) FROM item").fetchone()[0] == 2
print("PASS: replay, conflict detection, atomic rollback")

Ατομικότητα και σημεία συνέχισης

Η διαχείριση συναλλαγής με sqlite3 ολοκληρώνει τις αλλαγές όταν το σώμα εκτελεστεί χωρίς εξαίρεση και τις αναιρεί όταν προκύψει σφάλμα. Στην προβληματική παρτίδα, η προηγούμενη εισαγωγή με ID 3 πρέπει επίσης να αναιρεθεί. Έτσι δεν αποθηκεύονται μισές παρτίδες. Σε μόνιμο σύστημα, το σημείο από το οποίο θα συνεχίσει ο συγχρονισμός πρέπει να ενημερώνεται στην ίδια συναλλαγή με τις εγγραφές που έχουν πραγματικά αποθηκευτεί. Διαφορετικά, μία διακοπή μπορεί να οδηγήσει σε παράλειψη ή επανάληψη δεδομένων.

Πειράματα αποτυχίας

1. Επανεκτέλεσε τον ίδιο κωδικό με άλλη κατάσταση. Βεβαιώσου ότι εμφανίζεται σύγκρουση και δεν αλλάζει η παλιά εγγραφή.
2. Αντιστάθμισε τη σειρά των δύο προβληματικών εγγραφών. Έλεγξε πού αποτυγχάνει η εκτέλεση και ποια δεδομένα παραμένουν. Δοκίμασε επίσης αρνητικό κωδικό και άκυρη κατάσταση.
3. Πρόσθεσε έκδοση προέλευσης και όρισε ρητά ποιος μπορεί να ενημερώσει μία προηγούμενη καταχώριση. Σύγκρινε με την επικίνδυνη τυφλή αντικατάσταση.
4. Χρησιμοποίησε προσωρινή βάση σε αρχείο και έλεγξε διατήρηση έγκυρου σημείου συνέχισης μετά από κλείσιμο και επανεκκίνηση. Μη χρησιμοποιείς πραγματική βάση παραγωγής.

Παραδοτέα: PASS, πλήθος εγγραφών πριν και μετά, περιπτώσεις επανεκτέλεσης και κανόνας χειρισμού αντιφατικών δεδομένων. Το παράδειγμα καλύπτει μικρές παρτίδες, όχι ολόκληρη υποδομή ETL.