Die Architektur und Praxis der Synchronisierung von Meituan DB-Daten mit dem Data Warehouse

Hintergrund

Bei der Data-Warehouse-Modellierung werden die ursprünglichen Daten der Geschäftsschicht, die in keiner Weise verarbeitet wurden, als ODS-Daten (Operational Data Store) bezeichnet. Zu den gängigen ODS-Daten in Internetunternehmen zählen Geschäftsprotokolldaten (Log) und Geschäfts-DB-Daten (DB). Bei Business-DB-Daten ist das Sammeln von Geschäftsdaten aus relationalen Datenbanken wie MySQL und deren anschließender Import in Hive ein wichtiger Teil der Data Warehouse-Produktion.

Wie synchronisiere ich MySQL-Daten präzise und effizient mit Hive? Die häufig verwendete Lösung besteht darin, Daten stapelweise abzurufen und zu laden: Stellen Sie eine direkte Verbindung zu MySQL her, um Daten aus der Tabelle auszuwählen, speichern Sie sie dann als Zwischenspeicher in einer lokalen Datei und laden Sie die Datei schließlich in die Hive-Tabelle. Der Vorteil dieser Lösung liegt in ihrer einfachen Implementierung. Mit zunehmender Geschäftsentwicklung treten jedoch auch ihre Nachteile zutage:

• Leistungsengpass: Mit zunehmender Unternehmensgröße dauert der Datenfluss „Aus MySQL auswählen -> In lokaler Datei speichern -> In Hive laden“ immer länger, sodass die Zeitanforderungen der nachgelagerten Data Warehouse-Produktion nicht erfüllt werden können.
• Das Auswählen einer großen Datenmenge direkt aus MySQL hat enorme Auswirkungen auf MySQL, was leicht zu langsamen Abfragen führen und normale Dienste im Geschäftsbereich beeinträchtigen kann.
• Da die eigene Syntax von Hive keine SQL-Grundfunktionen wie „Aktualisieren“ und „Löschen“ unterstützt, kann es Daten, die in MySQL aktualisiert/gelöscht werden, nicht gut unterstützen.

Um diese Probleme vollständig zu lösen, haben wir uns schrittweise der technischen Lösung CDC (Change Data Capture) + Merge zugewandt, d. h. einer Lösung zur Echtzeit-Binlog-Sammlung + Offline-Verarbeitung von Binlog zur Wiederherstellung von Geschäftsdaten. Binlog ist das Binärprotokoll von MySQL, das alle Datenänderungen in MySQL aufzeichnet. Die Master-Slave-Synchronisierung des MySQL-Clusters selbst basiert auf Binlog.

In diesem Artikel wird hauptsächlich erläutert, wie DB-Daten aus zwei Blickwinkeln genau und effizient in das Data Warehouse eingegeben werden: Echtzeiterfassung von Binlog und Offlineverarbeitung von Binlog zum Wiederherstellen von Geschäftsdaten.

Gesamtarchitektur

Die Gesamtarchitektur ist in der Abbildung oben dargestellt. Im Hinblick auf die Echtzeit-Binlog-Sammlung haben wir das Open-Source-Projekt Canal von Alibaba übernommen, das dafür verantwortlich ist, Binlog in Echtzeit aus MySQL abzurufen und die entsprechende Analyse durchzuführen. Nachdem Binlog gesammelt wurde, wird es für die spätere Verwendung vorübergehend in Kafka gespeichert. Der gesamte Echtzeiterfassungsteil wird in der Abbildung durch den roten Pfeil angezeigt.

Die Offline-Binlog-Verarbeitung, wie in der Abbildung durch den schwarzen Pfeil dargestellt, stellt eine MySQL-Tabelle auf Hive mithilfe der folgenden Schritte wieder her:

• Das Open-Source-Projekt Camus von Linkedin wird verwendet, um die Binlog-Daten von Kafka stündlich nach Hive zu übertragen.
• Für jede ODS-Tabelle müssen Sie zunächst einmalig einen Snapshot erstellen, um die in MySQL vorhandenen Daten in Hive einzulesen. Der zugrundeliegende Prozess verwendet eine direkte Verbindung zu MySQL, um Daten auszuwählen.
• Für jede ODS-Tabelle wird jeden Tag eine Zusammenführung auf Basis der vorhandenen Daten und des durch das tägliche Inkrement generierten Binlogs durchgeführt, um die Geschäftsdaten wiederherzustellen.

Werfen wir einen Blick zurück auf die verschiedenen Probleme, die bei der im Hintergrund eingeführten Lösung zum Abrufen und Laden von Batchdaten aufgetreten sind. Warum kann diese Lösung die oben genannten Probleme lösen?

• Erstens wird Binlog im Streaming-Modus generiert. Durch die Echtzeiterfassung von Binlog wird ein Teil des Datenverarbeitungsbedarfs von der täglichen Stapelverarbeitung auf Echtzeit-Streams verteilt. Es wird deutliche Verbesserungen sowohl bei der Leistung als auch beim Zugriffsdruck auf MySQL geben.
• Zweitens zeichnet Binlog selbst die Art der Datenänderung auf (Einfügen/Aktualisieren/Löschen). Durch eine semantische Verarbeitung kann eine genaue Datenwiederherstellung erreicht werden.

Binlog-Echtzeitsammlung

Die Echtzeitsammlung von Binlog umfasst zwei Hauptmodule: Das eine ist CanalManager, das hauptsächlich für die Zuweisung von Sammlungsaufgaben, die Überwachung von Alarmen, die Metadatenverwaltung und das Andocken an externe abhängige Systeme verantwortlich ist; das andere sind Canal und CanalClient, die die Sammlungsaufgaben tatsächlich ausführen.

Wenn ein Benutzer eine Anfrage zum Sammeln von Binlogs für eine bestimmte Datenbank sendet, ruft CanalManager zunächst die entsprechende Schnittstelle der DBA-Plattform auf, um relevante Informationen über die MySQL-Instanz abzurufen, in der sich die Datenbank befindet, um dann die Maschine auszuwählen, die für das Sammeln von Binlogs am besten geeignet ist. Verteilen Sie dann die Sammlungsinstanz (Canal Instance) an den entsprechenden Canal-Server, nämlich CanalServer. Bei der Auswahl eines bestimmten CanalServers berücksichtigt CanalManager Faktoren wie Lastausgleich und Übertragung zwischen Rechenzentren und priorisiert Maschinen mit geringerer Last und Übertragung in derselben Region.

Nach Erhalt der Abholanforderung registriert CanalServer die Abholinformationen bei ZooKeeper. Der Registrierungsinhalt umfasst:

• Ein permanenter Knoten, der nach der Instanz benannt ist.
• Registrieren Sie unter dem permanenten Knoten einen temporären Knoten, der nach seiner eigenen IP:Port benannt ist.

Dies dient zwei Zwecken:

• Hohe Verfügbarkeit: Wenn CanalManager Instanzen verteilt, wählt er zwei CanalServer aus, einen als laufenden Knoten und den anderen als Standby-Knoten. Der Standby-Knoten überwacht die Instanz. Wenn der laufende Knoten ausfällt, verschwindet der temporäre Knoten und der Standby-Knoten übernimmt. Dadurch wird der Zweck der Notfallwiederherstellung erreicht.
• Interaktion mit CanalClient: Nachdem CanalClient den laufenden CanalServer erkannt hat, auf dem sich die Instanz befindet, für die er verantwortlich ist, stellt er eine Verbindung her, um die vom CanalServer gesendeten Binlog-Daten zu empfangen.

Das Abonnement von Binlog basiert auf der Granularität von MySQL DB und das Binlog einer DB entspricht einem Kafka-Thema. In der zugrunde liegenden Implementierung werden alle abonnierten DBs unter einer MySQL-Instanz von derselben Canal-Instanz verarbeitet. Dies liegt daran, dass Binlog auf der Granularitätsebene der MySQL-Instanz generiert wird. CanalServer verwirft die abbestellten Binlog-Daten und anschließend verteilt CanalClient das empfangene Binlog entsprechend der DB-Granularität an Kafka.

MySQL-Daten offline wiederherstellen

Nach Abschluss der Binlog-Sammlung besteht der nächste Schritt darin, Binlog zum Wiederherstellen von Geschäftsdaten zu verwenden. Das erste zu lösende Problem besteht darin, Binlog von Kafka mit Hive zu synchronisieren.

Kafka2Hive

Die Verwaltung der gesamten Kafka2Hive-Aufgabe erfolgt im ETL-Framework der Meituan-Datenplattform, einschließlich der Darstellung von Aufgabenprimitiven und Planungsmechanismen, die anderen ETLs ähneln. Die zugrunde liegende Schicht verwendet das Open-Source-Projekt Camus von LinkedIn und führt eine gezielte Sekundärentwicklung durch, um die eigentliche Datenübertragungsarbeit von Kafka2Hive abzuschließen.

Sekundärentwicklung von Camus

Das auf Kafka gespeicherte Binlog verfügt nicht über ein Schema, aber die Hive-Tabelle muss über ein Schema verfügen und das Design seiner Partitionen, Felder usw. muss eine effiziente nachgelagerte Nutzung ermöglichen. Die erste an Camus vorgenommene Änderung besteht darin, das Binlog auf Kafka in ein Format zu analysieren, das dem Zielschema entspricht.

Die zweite Transformation von Camus wurde durch das ETL-Framework von Meituan bestimmt. In unserem Aufgabenplanungssystem analysieren wir derzeit nur die Upstream- und Downstream-Abhängigkeiten von Aufgaben in derselben Planungswarteschlange. Abhängigkeiten über Planungswarteschlangen hinweg können nicht hergestellt werden. Im gesamten Prozess von MySQL2Hive muss die Kafka2Hive-Aufgabe einmal pro Stunde (stündliche Warteschlange) und die Merge-Aufgabe einmal täglich (Tageswarteschlange) ausgeführt werden. Der Start der Merge-Aufgabe muss strikt von der Fertigstellung der stündlichen Kafka2Hive-Aufgabe abhängig sein.

Um dieses Problem zu lösen, haben wir die Checkdone-Aufgabe eingeführt. Die Checkdone-Aufgabe ist eine tägliche Aufgabe, die hauptsächlich dafür verantwortlich ist, zu überprüfen, ob Kafka2Hive am Vortag erfolgreich abgeschlossen wurde. Bei erfolgreichem Abschluss wird die Checkdone-Aufgabe erfolgreich ausgeführt, sodass die nachgelagerte Merge-Aufgabe ordnungsgemäß gestartet werden kann.

Checkdone-Erkennungslogik

Wie erkennt Checkdone? Nachdem jede Kafka2Hive-Aufgabe die Datenübertragung erfolgreich abgeschlossen hat, ist Camus dafür verantwortlich, die Startzeit der Aufgabe im entsprechenden HDFS-Verzeichnis aufzuzeichnen. Checkdone scannt alle Zeitstempel des Vortages. Wenn der größte Zeitstempel 0:00 überschreitet, bedeutet dies, dass die Kafka2Hive-Aufgaben des Vortages erfolgreich abgeschlossen wurden und Checkdone die Erkennung abgeschlossen hat.

Da Camus selbst nur den Prozess des Lesens von Kafka und des anschließenden Schreibens von HDFS-Dateien abschließt, muss es außerdem auch das Laden der Hive-Partitionen abschließen, bevor es weiter unten abgefragt werden kann. Daher besteht der letzte Schritt der gesamten Kafka2Hive-Aufgabe darin, die Hive-Partitionen zu laden. Damit gilt die gesamte Aufgabe als erfolgreich ausgeführt.

Jede Kafka2Hive-Aufgabe ist dafür verantwortlich, ein bestimmtes Thema zu lesen und die Binlog-Daten in eine Tabelle unter der Bibliothek original_binlog zu schreiben, nämlich original_binlog.db in der vorherigen Abbildung, in der alle Binlogs gespeichert sind, die einer MySQL-Datenbank entsprechen.

Die obige Abbildung zeigt die Verzeichnisstruktur der Dateien auf HDFS, nachdem Kafka2Hive abgeschlossen ist. Wenn eine MySQL-Datenbank als Benutzer bezeichnet wird, wird das entsprechende Binlog in der Tabelle original_binlog.user gespeichert. Im Ready-Verzeichnis wird täglich die Startzeit aller erfolgreich ausgeführten Kafka2Hive-Aufgaben an diesem Tag zur Verwendung durch Checkdone gespeichert. Das Binlog jeder Tabelle ist in einer Partition organisiert. Beispielsweise wird das Binlog der Tabelle „userinfo“ in der Partition „table_name=userinfo“ gespeichert. Unter jeder Partition der ersten Ebene mit Tabellennamen werden die Partitionen der zweiten Ebene nach dt organisiert. Die Dateien xxx.lzo und xxx.lzo.index in der Abbildung speichern lzo-komprimierte Binlog-Daten.

Verschmelzen

Nachdem das Binlog erfolgreich gespeichert wurde, besteht der nächste Schritt darin, die MySQL-Daten basierend auf dem Binlog wiederherzustellen. Der Merge-Prozess führt zwei Dinge aus: Zunächst speichert er die am Tag generierten Binlog-Daten in der Delta-Tabelle und führt dann einen Primärschlüssel-basierten Merge mit den vorhandenen Bestandsdaten durch. Die Daten in der Delta-Tabelle sind die neuesten Daten des Tages. Wenn sich ein Datenelement an einem Tag mehrmals ändert, speichert die Delta-Tabelle nur die Daten nach der letzten Änderung.

Beim Zusammenführen von Delta-Daten mit Bestandsdaten ist ein eindeutiger Schlüssel erforderlich, um festzustellen, ob es sich um dieselben Daten handelt. Wenn in der Bestandstabelle und in der Delta-Tabelle dieselben Daten vorkommen, bedeutet dies, dass die Daten aktualisiert wurden und die Daten in der Delta-Tabelle als Endergebnis ausgewählt werden. Andernfalls bedeutet dies, dass keine Änderung aufgetreten ist und die Daten in der ursprünglichen Bestandstabelle als Endergebnis beibehalten werden. Die Ergebnisdaten der Zusammenführung werden durch Einfügen und Überschreiben in die Originaltabelle eingefügt, in der Abbildung also origindb.table.

Beispiel für einen Zusammenführungsprozess

Das folgende Beispiel veranschaulicht den Zusammenführungsprozess.

Die Datentabelle hat zwei Spalten: ID und Wert, wobei ID der Primärschlüssel ist. Beim Extrahieren von Deltadaten wird bei mehreren Aktualisierungen derselben Daten nur die zuletzt aktualisierte Version ausgewählt. Daher zeichnet die Delta-Tabelle für die Daten mit der ID=1 den zuletzt aktualisierten Wert „Wert=120“ auf. Nachdem die Delta-Daten und die vorhandenen Daten zusammengeführt wurden, werden im Endergebnis neue Daten eingefügt (ID = 4), zwei Daten werden aktualisiert (ID = 1 und ID = 2) und ein Datenwert bleibt unverändert (ID = 3).

Standardmäßig verwenden wir den Primärschlüssel der MySQL-Tabelle als eindeutigen Schlüssel für diese Duplikatserkennung. Das Unternehmen kann je nach tatsächlichen Bedingungen auch einen anderen eindeutigen Schlüssel als MySQL konfigurieren.

Oben wird die Gesamtarchitektur der Binlog-basierten Datenerfassung und ODS-Datenwiederherstellung vorgestellt. Im Folgenden werden die tatsächlichen Geschäftsprobleme, die wir gelöst haben, hauptsächlich aus zwei Blickwinkeln vorgestellt.

Übung 1: Unterstützung für Sharding

Mit der Ausweitung des Geschäftsumfangs verfügt MySQL über immer mehr Shard-Datenbanken und -Tabellen, und die Anzahl der Shard-Tabellen liegt bei vielen Unternehmen im Tausenderbereich. Im Allgemeinen müssen Datenentwickler diese Daten zur Analyse zusammenfassen. Wenn wir jede Shard-Tabelle manuell synchronisieren und sie dann auf Hive aggregieren müssen, sind die Kosten für uns kaum zu akzeptieren. Daher müssen wir die Aggregation der Untertabellen auf der ODS-Ebene abschließen.

Erstens unterstützen wir beim Sammeln von Binlogs in Echtzeit das Schreiben von Binlogs aus verschiedenen DBs in dasselbe Kafka-Thema. Bei der Beantragung einer Binlog-Sammlung können Benutzer mehrere physische Datenbanken gleichzeitig unter derselben Geschäftslogik auswählen. Durch die Aggregation auf der Ebene der Binlog-Sammlung werden die Binlogs aller Shards in dieselbe Hive-Tabelle geschrieben, sodass der Downstream beim Zusammenführen nur eine Hive-Tabelle lesen muss.

Zweitens unterstützt die Konfiguration der Merge-Aufgabe die Übereinstimmung mit regulären Ausdrücken. Durch Konfigurieren eines regulären Ausdrucks, der den Benennungsregeln der Geschäftspartitionstabellen entspricht, kann die Merge-Aufgabe erkennen, welche MySQL-Tabellen-Binlogs aggregiert werden müssen, und so die Daten der entsprechenden Partition zur Ausführung auswählen.

Auf diese Weise wird durch die Arbeit auf zwei Ebenen die Zusammenführung von Unterdatenbanken und Untertabellen auf der ODS-Ebene abgeschlossen.

Hier gibt es eine technische Optimierung. Bei der Ausführung von Kafka2Hive haben wir die Tabellennamen gemäß den Partitionierungsregeln für Geschäftstabellen verarbeitet und die physischen Tabellennamen in logische Tabellennamen konvertiert. Beispielsweise wird der Tabellenname userinfo123 in userinfo konvertiert und seine Binlog-Daten werden in der Partition table_name=userinfo der Tabelle original_binlog.user gespeichert. Dies geschieht, um den zugrunde liegenden Druck zu verhindern, der durch zu viele kleine HDFS-Dateien und Hive-Partitionen verursacht wird.

Übung 2: Unterstützung für Löschereignisse

Löschvorgänge sind in MySQL sehr verbreitet. Da Hive Löschen nicht unterstützt, müssen Sie eine „umgehbare“ Methode verwenden, wenn Sie die in MySQL gelöschten Daten in Hive löschen möchten.

Für den Merge-Prozess, der das Delete-Ereignis verarbeiten muss, werden die folgenden zwei Schritte verwendet:

• Extrahieren Sie zunächst die Daten, bei denen das Löschereignis aufgetreten ist. Da Binlog selbst den Ereignistyp aufzeichnet, ist dieser Schritt einfach durchzuführen. Führen Sie einen Left Outer Join für den Primärschlüssel der vorhandenen Daten (Tabelle A) und der gelöschten Daten (Tabelle B) durch. Wenn alle Daten auf beiden Seiten verknüpft werden können, bedeutet dies, dass die Daten gelöscht wurden. Daher sind die Daten, die dem Datensatz der Tabelle B im Auswahlergebnis entsprechen, der NULL ist, die Daten, die beibehalten werden sollten.
• Führen Sie dann für die oben erhaltenen gespeicherten Daten eine reguläre Zusammenführung gemäß dem oben beschriebenen Verfahren durch.

Ausblick

Als Grundlage für die Data Warehouse-Produktion deckt der von der Meituan-Datenplattform bereitgestellte Binlog-basierte MySQL2Hive-Dienst grundsätzlich alle Geschäftsbereiche innerhalb von Meituan ab. Er kann derzeit die Datensynchronisierungsanforderungen der meisten Unternehmen erfüllen und eine genaue und effiziente Lagerung von DB-Daten realisieren. Bei der zukünftigen Entwicklung werden wir uns auf die Lösung des Einzelpunktproblems von CanalManager konzentrieren und eine datencenterübergreifende Notfallwiederherstellungsarchitektur aufbauen, um die Geschäftsentwicklung stabiler zu unterstützen.

Dieser Artikel stellt die Architektur dieses Dienstes hauptsächlich aus zwei Blickwinkeln vor: Binlog-Streaming-Sammlung und Binlog-basierte ODS-Datenwiederherstellung, und stellt einige typische Probleme und Lösungen vor, die uns in der Praxis begegnet sind. Ich hoffe, dass dies anderen Entwicklern als Referenz dienen kann, und jeder ist herzlich eingeladen, mit uns zu kommunizieren.

Zusammenfassen

Das Obige ist der vollständige Inhalt dieses Artikels. Ich hoffe, dass der Inhalt dieses Artikels einen gewissen Lernwert für Ihr Studium oder Ihre Arbeit hat. Vielen Dank für Ihre Unterstützung von 123WORDPRESS.COM. Wenn Sie mehr darüber erfahren möchten, schauen Sie sich bitte die folgenden Links an