cumulative table design

Kumulatives Tischdesign

Das kumulative Tabellendesign ist ein äußerst leistungsstarkes Data-Engineering-Tool, das alle Dateningenieure kennen sollten.

Dieses Design erzeugt Tabellen, die effiziente Analysen für beliebig große Zeiträume (bis zu Tausenden von Tagen) ermöglichen können

Hier ist ein Diagramm des High-Level-Pipeline-Designs für dieses Muster:

Wir erstellen zunächst unsere tägliche Metriktabelle, die den Kern unserer Entität bildet. Diese Daten stammen aus allen Ereignisquellen, die wir vorgelagert haben.

Nachdem wir unsere täglichen Metriken haben, führen wir FULL OUTER JOIN der gestrigen kumulativen Tabelle mit den heutigen Tagesdaten durch und erstellen unsere Metrik-Arrays für jeden Benutzer. Dadurch können wir den neuen Verlauf einbinden, ohne ihn vollständig scannen zu müssen. (ein großer Leistungsschub)

Mit diesen Metrik-Arrays können wir problemlos Anfragen zum Verlauf aller Benutzer beantworten, indem wir Dinge wie ARRAY_SUM verwenden, um jede gewünschte Metrik in dem vom Array zugelassenen Zeitrahmen zu berechnen.

Je länger der Zeitrahmen Ihrer Analyse ist, desto kritischer wird dieses Muster!!

Beispiel: Benutzeraktivität und -engagement kumuliert

Die gesamte Abfragesyntax verwendet Presto/Trino-Syntax und -Funktionen. Dieses Beispiel müsste für andere SQL-Varianten angepasst werden!

Wir verwenden die Daten:

• 01.01.2022 wie heute in Bezug auf den Luftstrom ist dies {{ ds }}
• 31.12.2021 wie gestern in Bezug auf Airflow-Vorlagen ist dies {{ yesterday_ds}}

In diesem Beispiel untersuchen wir, wie dieses Design erstellt wird, um täglich, wöchentlich und monatlich aktive Benutzer sowie die Likes, Kommentare und Shares der Benutzer zu berechnen.

Unsere Quelltabelle ist in diesem Fall events .

• Ein Benutzer ist an einem bestimmten Tag aktiv, wenn er für diesen Tag ein Ereignis generiert.
• Die Tabelle hat event_type , der like , comment , share oder view ist

Es ist verlockend zu glauben, dass die Lösung für dieses Problem darin besteht, so etwas wie eine Pipeline zu betreiben

    SELECT 
        COUNT(DISTINCT user_id) as num_monthly_active_users,
        COUNT(CASE WHEN event_type = 'like' THEN 1 END) as num_likes_30d,
        COUNT(CASE WHEN event_type = 'comment' THEN 1 END) as num_comments_30d,
        COUNT(CASE WHEN event_type = 'share' THEN 1 END) as num_shares_30d,
        ...
    FROM events
    WHERE event_date BETWEEN DATE_SUB('2022-01-01', 30), AND '2022-01-01'

Das Problem dabei ist, dass wir täglich 30 Tage lang Ereignisdaten scannen, um diese Zahlen zu ermitteln. Eine ziemlich verschwenderische, aber dennoch einfache Pipeline. Es sollte eine Möglichkeit geben, die Ereignisdaten nur einmal zu scannen und mit den Ergebnissen der letzten 29 Tage zu kombinieren, oder? Können wir eine Datenstruktur erstellen, in der ein Datenwissenschaftler unsere Daten abfragen und leicht ermitteln kann, wie viele Aktionen ein Benutzer in den letzten N Tagen ausgeführt hat?

Dieses Design ist mit nur 3 Schritten ziemlich einfach:

Der tägliche Tischschritt

• In diesem Schritt fassen wir nur die Ereignisse von heute zusammen, um zu sehen, wer täglich aktiv ist. Das Tabellenschema finden Sie hier
• Diese Abfrage ist hier ziemlich einfach und unkompliziert
  • GROUP BY user_id und zähle sie dann als täglich aktiv, wenn sie irgendwelche Ereignisse haben
  • Wir fügen ein paar COUNT(CASE WHEN event_type = 'like' THEN 1 END) -Anweisungen hinzu, um auch die Anzahl der täglichen Likes, Kommentare und Shares zu ermitteln

Der Kumulierungsschritt

• Das Tabellenschema für diesen Schritt finden Sie hier
• Die Abfrage für diesen Schritt ist deutlich komplexer. Es ist hier
  • In diesem Schritt übernehmen wir die heutigen Daten aus der Tagestabelle und die gestrigen Daten aus der kumulierten Tabelle
  • Wir FULL OUTER JOIN dieser beiden Datensätze auf today.user_id = yesterday.user_id durch
  • Wenn ein Benutzer ganz neu ist, ist er nicht in den Daten von gestern enthalten. Wenn ein Benutzer heute nicht aktiv war, ist er auch nicht in den heutigen Daten enthalten
  • Wir müssen also COALESCE(today.user_id, yesterday.user_id) as user_id verwenden, um den Überblick über alle Benutzer zu behalten
  • Als nächstes wollen wir die Spalte activity_array erstellen. Wir möchten nur, dass activity_array die Daten der letzten 30 Tage speichert
    • Wir prüfen also, ob CARDINALITY(activity_array) < 30 um zu verstehen, ob wir den heutigen Wert einfach am Anfang des Arrays hinzufügen können oder ob wir ein Element vom Ende des Arrays abschneiden müssen, bevor wir den heutigen Wert am Anfang hinzufügen
    • Wir müssen COALESCE(t.is_active_today, 0) ausführen, um Nullwerte in das Array einzufügen, wenn ein Benutzer nicht aktiv ist
  • Wir folgen einem sehr ähnlichen Muster wie wir activity_array erstellt haben, aber auch für Likes, Kommentare und Shares!
  • Nachdem wir alle unsere Metrik-Arrays erstellt haben, ist die Berechnung der wöchentlichen und monatlichen Aktivität ziemlich einfach
    • CASE WHEN ARRAY_SUM(activity_array) > 0 THEN 1 ELSE 0 END gibt uns monatliche Aktiven, da wir die Array-Größe auf 30 begrenzen
    • CASE WHEN ARRAY_SUM(SLICE(activity_array, 1, 7)) > 0 THEN 1 ELSE 0 END gibt uns wöchentliche Aktivität, da wir nur die ersten 7 Elemente des Arrays prüfen (also die letzten 7 Tage)
  • Es ist auch einfach, die Anzahl der Likes, Kommentare und Shares der letzten 7 Tage zu summieren
    • ARRAY_SUM(SLICE(like_array, 1, 7)) as num_likes_7d gibt uns die Anzahl der Likes an, die dieser Benutzer in den letzten 7 Tagen gemacht hat
    • ARRAY_SUM(like_array) as num_likes_30d gibt uns die Anzahl der Likes an, die dieser Benutzer in den letzten 30 Tagen gemacht hat, seit das Array auf diese Größe festgelegt ist

Der DAG-Schritt

• Den Beispiel-DAG finden Sie hier
• Wichtige Dinge, an die sich die DAG erinnern sollte, sind:
  • Kumulative DAGs sollten immer depends_on_past: True sein
  • Der tägliche Aggregationsschritt muss dem Kumulierungsschritt vorgeschaltet sein