Datenstabilität mit Python: Wie man selbst die kleinsten Änderungen erfasst

Als Dateningenieur ist es fast immer die sicherste Option, Datenpipelines alle X Minuten auszuführen. So kannst du nachts gut schlafen, in dem Wissen, dass du keine Updates aus der Quelle verpasst hast.

Leider ist dieser Ansatz nicht wirklich vorteilhaft, wenn man die hohe Belastung der Infrastruktur und die Rechnung am Monatsende berücksichtigen muss. Ganz oft ist es nur wirklich notwendig, sie auszuführen, wenn es Änderungen in der Quelle gegeben hat. Außerdem bedeutet ein einfaches erneutes Ausführen, dass wir nicht wirklich wissen, wann die echten Updates in der Quelle stattgefunden haben. Eine Möglichkeit, zu überprüfen, ob 2 Dateien identisch sind, besteht darin, ihre Hashes zu vergleichen, was mit einer einfachen Funktion erfolgen kann, die unten gezeigt wird:

import hashlib

def calculate_file_hash_chunked(file_path: str) -> str:
"""
Berechne den Hash einer Datei mit Chunk-basiertem Ansatz
Aufgrund des kürzeren Hashs und damit eines besser lesbaren Hashs wird md5 verwendet.

:param file_path: Der Pfad zur CSV-Datei.
:return: Der Hash der Datei als hexadezimale Zeichenkette.
"""
hash_obj = hashlib.md5()
with open(file_path, 'rb') as file:
for chunk in iter(lambda: file.read(4096), b""):
hash_obj.update(chunk)
return hash_obj.hexdigest()

Hashes sind im Allgemeinen alphanumerische Zeichenfolgen, die einen deterministischen Algorithmus verwenden, um ein eindeutiges Ergebnis eines Eingangs zu berechnen, was einen der Gründe dafür darstellt, dass sie verwendet werden können. Ein deterministischer Algorithmus bedeutet in diesem Kontext, dass es keine externen Eingaben gibt und keine Zufälligkeit existiert, außer für die bereitgestellte Eingabe. Hier siehst du ein Beispiel für einen "kürzeren" MD5-Hash; `938c2cc0dcc05f2b68c4287040cfcf71`.

Wir könnten jedoch feststellen, dass dieser Ansatz unzureichend konsistent ist, da wir möglicherweise nicht wissen, wie die Dateien in der Quelle erzeugt werden. Die Tatsache, dass das Verschieben einer einzigen Zeile um 1 Zeile nach oben/unten einen völlig anderen Hash erzeugt, macht es schwierig zu wissen, ob sich die Daten darin geändert haben, da wir tatsächlich wissen möchten, ob sich die Daten geändert haben und nicht die Datei selbst.

Hat sich die Daten geändert oder sind die Zeilen einfach umsortiert?

Pandas

Um dies zu lösen, ist eine einfache pandas DataFrame-Methode normalerweise ausreichend, um Änderungen zu identifizieren. Wenn wir die Daten in einen DataFrame laden, dann die Spalten sortieren und anschließend alle Zeilen sortieren, können wir mit Sicherheit sagen, dass sich die Daten nicht geändert haben, indem wir alles zusammen hashen.

Eine wichtige Erwähnung sollte der Stabilität der Sortierung gewidmet werden, da es ein wichtiges Merkmal ist, die gleichen Ergebnisse bei der Ausführung zu erhalten. Eine stabile Sortierung ist eine Sortierung, die die Reihenfolge von Elementen mit identischen Werten beibehält, wie sie vor der Sortierung waren. Allerdings gehen wir davon aus, dass nur die Daten innerhalb des DataFrames von Bedeutung sind und dass der DataFrame einfach umsortiert werden könnte, sodass die Stabilität der Sortierung keine Rolle spielt. Daher ist es unabhängig von der Stabilität der Sortierung der Umstand, dass wir die Zeilen nacheinander sortieren, eine Option, die nicht berücksichtigt werden muss.

import pandas as pdfrom typing import Optionaldef sort_dataframe(    df: pd.DataFrame,    sort_rows: Optional[bool] = True,    sort_columns: Optional[bool]

Dies ermöglicht es uns bereits, zwei mehrdeutige DataFrames einfach zu vergleichen, wie unten zu sehen:

data1 = {    'A': [True, False, True],    'B': [1, 2, 3],    'C': ['foo', 'bar', 'baz']}data2 = {    'C': ['baz', 'foo', 'bar'],    'B': [3, 1, 2],    'A': [True, True, False]

Wie oben zu sehen ist, werden boolesche Werte sortiert, wobei die Spalte B den entscheidenden Faktor dafür bildet, welcher Wert zuerst kommt, da sie den gleichen Wert in Spalte A haben.

Hashing

Allerdings kann es oft ziemlich mühsam, zeitaufwändig und ineffizient sein, Zugriff auf den vorherigen DataFrame zu erhalten. Was mehr Sinn macht, ist einfach den gesamten DataFrame zu hashen und das Ergebnis irgendwo zu speichern, wo es leicht abgerufen werden kann.

Das obige Beispiel lädt es direkt in den Speicher und hasht es im Feather-Format, welches deterministisch ist. Wenn die Hashes von zwei separaten Dateien, oder in diesem Fall, DataFrames, die gleichen sind, bedeutet dies, dass entweder die beiden Dateien identisch sind oder ein Hash-Kollision aufgetreten ist. Da wir versuchen zu bestimmen, ob zwei Objekte identisch sind, können wir die viel wahrscheinlichere Wahrscheinlichkeit annehmen ( ± 1 in 10²⁹ ), dass die beiden Datensätze identisch sind.

Dieser Ansatz funktioniert auch für andere Formate wie Parquet, wenn sie ebenfalls in Pandas geladen werden, da die Kernprozesse des Sortierens und Hashens gleich bleiben.

Bleib dran für den nächsten Beitrag über das Hashing von Datendateien, in dem verschiedene Möglichkeiten zum Hashing hervorgehoben werden und die überlegene Geschwindigkeit von Polars demonstriert wird.