1. Einführung in AWS MSK und Datenpipelines
Daten sind das neue Öl, und genau wie Öl müssen sie raffiniert und verarbeitet werden, um nützlich zu sein. In der datengesteuerten Welt von heute müssen Unternehmen riesige Datenmengen verarbeiten, um fundierte Entscheidungen treffen zu können. Datenpipelines sind die Lösung für dieses Problem. Eine Datenpipeline ist eine Reihe miteinander verbundener Schritte, die Daten verarbeiten und von ihrer Rohform in ein Format umwandeln, das analysiert und genutzt werden kann.
Dieser Blog soll allen helfen, die die Integration von Datenpipelines mit Kafka und Kafka Connect in AWS unter Verwendung von MSK und MSK Connect Managed Services planen oder gerade dabei sind, und gibt einen kleinen Einblick anhand von Konfigurationsbeispielen und Fehlern, die mir bei meinem Proof of Concept begegnet sind.
2. Was ist eine Datenpipeline?
Eine Datenpipeline ist eine Reihe von Verfahren, mit denen Rohdaten in ein für die Analyse geeignetes Format umgewandelt werden. Datenaufnahme, -bereinigung, -umwandlung und -speicherung sind einige der Schritte, die durchgeführt werden können. Je nachdem, wie viele Daten betroffen sind und wie viel Verarbeitung erforderlich ist, kann eine Datenpipeline einfach oder komplex sein. Das Hauptziel einer Datenpipeline besteht darin, die Daten für die Analyse vorzubereiten.
2.1 Komponenten der Datenpipeline
Für die Erstellung von Datenpipelines werden zahlreiche Technologien verwendet, z. B. Cloud-basierte Plattformen, Datenbanksysteme und Programmiersprachen. Die grundlegenden Komponenten einer Datenpipeline sind:
- Datenquelle: Die Suche nach der Datenquelle ist der erste Schritt. Dies kann ein Objektspeichersystem, eine Datenbank, eine Datei oder eine API sein.
- Die Daten werden anschließend während der Datenaufnahme in die Datenpipeline eingegeben. In dieser Phase werden die Daten aus ihrer Quelle extrahiert und in die Pipeline geladen.
- Die Datenbereinigung ist notwendig, sobald die Daten übernommen wurden. Der Prozess der Datenbereinigung umfasst die Beseitigung von Duplikaten, Fehlern oder Diskrepanzen.
- Datenumwandlung: Nach der Bereinigung werden die Daten in ein analysefähiges Format gebracht. In diesem Schritt werden die Daten strukturiert, es werden neue Variablen erstellt oder die Daten werden aggregiert.
- Speicherung der Daten: Die konvertierten Daten müssen in einem letzten Schritt vor der Analyse in einer Datenbank oder Datei gespeichert werden.
2.2 Arten von Datenpipelines
Abhängig von der Art der Daten, den Verarbeitungsanforderungen und der Art der erforderlichen Analyse gibt es verschiedene Arten von Datenpipelines, die jeweils für einen bestimmten Bedarf erstellt werden. Unternehmen können die beste Datenpipeline für ihre Bedürfnisse bei der Datenverarbeitung auswählen, wenn sie sich über die verschiedenen verfügbaren Typen im Klaren sind. Ich werde zwei der gängigsten Typen erwähnen.
- Batch-Verarbeitung: Bei dieser Art von Datenpipeline werden Daten in vorgegebenen Intervallen oder Stapeln verarbeitet, in der Regel einmal pro Tag oder einmal pro Woche. Für die Verarbeitung großer Datenmengen, die keine Echtzeitverarbeitung erfordern, sind Batch-Pipelines sehr effektiv. Anwendungen wie Business Intelligence, Reporting und Data Warehousing nutzen häufig Batch-Pipelines.
- Streaming-Pipeline: Die Verarbeitung von Daten in Echtzeit oder nahezu in Echtzeit erfolgt über Datenstrom-Pipelines. Stream-Pipelines eignen sich für Anwendungen wie Betrugserkennung, Echtzeitanalysen und Echtzeitüberwachung, bei denen Daten in Echtzeit analysiert werden müssen. Technologien wie Apache Kafka, Apache Flink und Apache Spark Streaming werden häufig in Stream-Pipelines eingesetzt.
Beide Typen können entweder ETL oder ELT durchführen.
- Extrahieren, Transformieren, Laden (ETL) Pipeline: ETL-Pipelines werden verwendet, um Daten aus verschiedenen Quellen zu extrahieren, sie in ein für die Analyse geeignetes Format umzuwandeln und sie dann in eine Zieldatenbank oder ein Data Warehouse zu laden. ETL-Pipelines werden häufig für Datenintegration, Datenmigration und Data Warehousing verwendet.
- Extrahieren, Laden, Transformieren (ELT) Pipeline: ELT-Pipelines ähneln ETL-Pipelines, aber die Umwandlung der Daten erfolgt, nachdem sie in die Zieldatenbank oder das Data Warehouse geladen wurden. ELT-Pipelines werden häufig für die Big-Data-Verarbeitung verwendet, bei der die Umwandlung der Daten in der Zieldatenbank oder im Data Warehouse mithilfe von verteilten Verarbeitungstechnologien wie Apache Hadoop, Apache Spark und Apache Hive erfolgt.
In diesem Blog werden wir uns auf den Streaming-Pipeline-Typ. Wir werden die verwalteten Dienste verwenden AWS MSK und AWS MSK Verbinden.
3. AWS MSK und AWS MSK Connect
3.1 AWS MSK
Beim Aufbau von Echtzeit-Datenpipelines und Streaming-Anwendungen wird das beliebte Open-Source-Streaming-Framework Apache Kafka verwendet. Kafka eignet sich am besten für Szenarien wie Echtzeitanalysen, Datenintegration und ereignisgesteuerte Anwendungen, da es einen hohen Durchsatz und geringe Latenzzeiten bei der Datenverarbeitung bietet.
Der Aufbau und Betrieb von Kafka-basierten Cloud-Anwendungen ist mit AWS MSK (Amazon Managed Streaming for Apache Kafka), einem vollständig verwalteten Service, einfach. Sie können Skalierbarkeit, Zuverlässigkeit und Leistung schnell und einfach verwalten und gleichzeitig Sicherheitseinstellungen für Kafka-Cluster mit Amazon MSK konfigurieren.
Wie funktioniert AWS MSK?
AWS MSK lässt sich mit anderen AWS-Services wie Amazon S3, Amazon Kinesis und Amazon EC2 integrieren, um eine vollständige Streaming-Datenplattform bereitzustellen.
Amazon Managed Streaming für Apache Kafka unterstützt die neueste Version von Kafka und bietet automatische Upgrades, Überwachung und Verwaltung von Kafka-Clustern. AWS MSK lässt sich auch in AWS-Sicherheitsservices wie AWS Identity and Access Management (IAM), AWS Key Management Service (KMS) und Amazon VPC integrieren, um eine sichere und konforme Datenverarbeitung zu gewährleisten.
3.2 AWS MSK Connect: Vereinfachung von Kafka-Datenpipelines
AWS MSK Connect ist eine neue Funktion von AWS MSK, die den Prozess der Erstellung und Verwaltung von Datenpipelines mit Kafka vereinfacht. Mit AWS MSK Connect können Sie problemlos Daten zwischen Kafka und anderen Datenspeichersystemen wie Amazon S3, Amazon Redshift, Amazon Elasticsearch und Amazon Relational Database Service (AWS RDS) verschieben.
AWS MSK Connect bietet eine einfache und skalierbare Möglichkeit zum Aufbau von Datenpipelines, indem Sie Konnektoren definieren können, die Daten zwischen Kafka und anderen Datensystemen übertragen. AWS MSK Connect unterstützt eine breite Palette von Konnektoren, einschließlich JDBC, S3, Elasticsearch und HTTP, unter anderem.
3.3. Vorteile und Nachteile der Verwendung von AWS MSK und AWS MSK Connect
AWS MSK und AWS MSK Connect sind leistungsstarke Tools, mit denen Sie Ihre Datenverarbeitungsabläufe optimieren und Kafka-basierte Datenpipelines in der Cloud erstellen und verwalten können. Wie jede Technologie haben sie jedoch sowohl Vorteile als auch Schattenseiten.
Vorteile der Verwendung von AWS MSK und AWS MSK Connect:
- Skalierbarkeit: AWS MSK ist ein vollständig verwalteter Service, der eine automatische Skalierung von Kafka-Clustern auf der Grundlage der Arbeitslast bietet, sodass Sie große Datenmengen verarbeiten und den Durchsatz nach Bedarf erhöhen können.
- Hohe Verfügbarkeit: AWS MSK bietet ein automatisches Failover und eine Replikation von Daten, um sicherzustellen, dass Ihre Daten immer verfügbar und dauerhaft sind.
- Vereinfachte Überwachung: Die Überwachung kann durch AWS Cloudwatch erfolgen oder Sie können Prometheus-Metriken abrufen, indem Sie "Enable open monitoring with Prometheus" aktivieren.
- Integration mit anderen AWS-Services: AWS MSK lässt sich in andere AWS-Services wie Amazon S3, Amazon RDS und andere integrieren, sodass Sie End-to-End-Pipelines für Streaming-Daten erstellen können, die Daten in Echtzeit verarbeiten, speichern und analysieren.
Nachteile der Verwendung von AWS MSK und AWS MSK Connect:
- Kosten: Wenn große Datenmengen verarbeitet werden oder ein hoher Durchsatz erforderlich ist, können AWS MSK und AWS MSK Connect kostspielig sein.
- Lernkurve: Die Einrichtung und Verwendung von AWS MSK und AWS MSK Connect kann eine Herausforderung sein und erfordert ein tiefes Verständnis von Kafka und anderen Datenverarbeitungstechnologien.
- Bindung an den Anbieter: Je nach Ihrer aktuellen Konfiguration kann sich der Wechsel zu einem anderen Anbieter als recht schwierig erweisen, da AWS MSK und AWS MSK Connect proprietäre Dienste sind, die von Amazon Web Services (AWS) bereitgestellt werden.
- Begrenzte Anpassungsmöglichkeiten: Da es sich bei AWS MSK und AWS MSK Connect um vollständig verwaltete Services handelt, haben Sie kaum Möglichkeiten, die Einrichtung und Anpassung von Kafka-Clustern und Verbindungen zu ändern.
4. Die Einrichtung
Datenpipelines können je nach Anwendungsfall mit verschiedenen Technologien, Tools und Programmiersprachen aufgebaut werden. Der Einfachheit halber werde ich Ihnen jedoch ein hypothetisches Layout geben.
Infrastrukturkomponenten:
- AWS VPC
- 3x Availability Zones (us-east-1a, us-east-1b, us-east-1c)
- 3x private Teilnetze
- 3x NAT-Gateways
- 2x RDS (Aurora oder einzelne RDS-Instanz)
- 1x MSK Cluster - 3x Makler und 3x Zoowärter
- 2x MSK Steckverbinder
- 2x MSK Connect Custom Plugins (eines für jede Art von Stecker-Plugin)
- 1x S3 Bucket für die gezippten Plugins - Debezium und JDBC Sink Connector für Confluent
Wie aus dem obigen Diagramm ersichtlich, haben wir eine VPC mit 3 privaten Subnetzen, jedes in einer anderen Verfügbarkeitszone mit eigenem NAT-Gateway (dies wird von den MSK Konnektoren für die Internetkonnektivität benötigt). Die MSK Connect Custom Plugins werden von unseren Konnektoren verwendet, um die gezippten Kafka-Connect-Plugins zu importieren - Debezium und JDBC Sink Connector für Confluent.
Der Quell-Connector streamt alle Zeilenänderungen aus der Quelldatenbank, lädt sie in ein Kafka-Thema hoch und der Sink-Connector konsumiert und schreibt sie in die Zieldatenbank.
Im Folgenden führe ich jede Komponente auf, nenne Details zu ihr und die Probleme oder Fehler, auf die ich gestoßen bin, sowie deren Lösungen.
4.1 PostgreSQL-Quelldatenbank
Dies wird die Datenbank unseres Mikrodienstes sein, die den Namen APP1von der aus wir alle Zeilenänderungen in der Datenbank als Nachrichten an ein Kafka-Thema streamen werden. Diese Methode wird auch als Änderungsdatenerfassung(CDC). Anschließend werden diese Nachrichten über den Kafka-Sink-Connector in die aggregierte Datenbank geschrieben.
Damit dies geschehen kann, müssen wir die logische Replikation der RDS-Instanz einschalten.
Dies kann durch die Erstellung einer "Paramater-Gruppe" erfolgen, wenn Sie eine Einzel-RDS-Instanz oder eine "DB Cluster Paramater Group", wenn Sie einen Aurora-Cluster. In der Parametergruppe müssen Sie "rds.logical_replication=1" setzen und anwenden. Danach starten Sie Ihre Datenbank neu.
Greifen Sie auf Ihre Datenbank zu und führen Sie "show wal_level;".
Erwartetes Ergebnis:
app1_db=> show wal_level;
2 wal_level
3-----------
4 logical
5(1 row)
Wenn die Anzeige "Replik" anzeigt, wurde die Parametergruppe nicht erfolgreich angewendet.
4.1 PostgreSQL-Zieldatenbank
Für die Zieldatenbank müssen keine zusätzlichen Konfigurationen vorgenommen werden. In unserem Fall wird dies die "Aggregierte Datenbank" sein.
5. Schema-Register
Aus der offiziellen confluent-Dokumentation - "Schema Registry bietet eine Bereitstellungsschicht für Ihre Metadaten. Sie bietet eine RESTful-Schnittstelle zum Speichern und Abrufen von Avro-Schemata. Sie speichert eine versionierte Historie aller Schemata, bietet mehrere Kompatibilitätseinstellungen und ermöglicht die Entwicklung von Schemata entsprechend der konfigurierten Kompatibilitätseinstellung. Es bietet Serialisierer, die sich in Kafka-Clients einfügen und die Speicherung und den Abruf von Schemas für Kafka-Nachrichten, die im Avro-Format gesendet werden, übernehmen."
- Die Quelle finden Sie HIER.
- Für unsere Schemaregistrierung werden wir die Schemaregistrierung von Confluent verwenden.
- Die offizielle Dokumentation finden Sie HIER.
Für die Serialisierung von Daten stehen mehrere Optionen zur Auswahl, aber wir werden uns nur auf eine konzentrieren - Avro.
Avro ist ein von der Apache Software Foundation entwickeltes Format zur Serialisierung von Daten. Es ist ein kompaktes und effizientes Binärformat, das eine schnelle Serialisierung und Deserialisierung von Daten ermöglicht. Avro bietet Funktionen wie Schema-Evolution, dynamische Typisierung und Unterstützung für komplexe Datenstrukturen, was es zu einer beliebten Wahl für Daten-Pipelines und verteilte Systeme macht.
Einer der Hauptvorteile von Avro ist die Unterstützung für Schema-Evolution. Da sich Datenstrukturen im Laufe der Zeit ändern, ermöglicht Avro die Evolution von Datenschemata, ohne dass Änderungen am Code oder an den Datendateien erforderlich sind. Dies erleichtert die Verwaltung von Datenpipelines und verteilten Systemen, auf denen unterschiedliche Versionen desselben Datenschemas laufen können.
Avro ist außerdem in hohem Maße interoperabel und unterstützt eine Vielzahl von Programmiersprachen und Plattformen. Dies macht es einfach, Avro in bestehende Systeme zu integrieren und Daten zwischen verschiedenen Teilen eines verteilten Systems auszutauschen.
Darüber hinaus reduziert Avro auch die Größe der Nachrichten in einem Kafka-Thema, was bei Millionen und Abermillionen von Nachrichten in Bezug auf Latenzzeit und Speichereffizienz ein entscheidender Faktor sein kann.
Einfaches Diagramm:
Alternative Schemaregister:
6. MSK Cluster Konfiguration
AWS MSK wird mit Standardkonfigurationen geliefert, aber Sie können auch benutzerdefinierte Konfigurationen sowohl für die Kafka-Broker als auch für die Kafka-Zookeeper erstellen. Die benutzerdefinierten Konfigurationen verfügen über eine Versionierung und können auf mehrere MSK-Cluster angewendet werden. Kafka-Konfigurationen können sehr komplex sein und variieren auf der Grundlage verschiedener Faktoren wie der durchschnittlichen Nachrichtengröße, der Anzahl der Nachrichten pro Sekunde, Aufbewahrungsrichtlinien, Replikation, Verzögerung und vielen anderen.
- Sehen Sie sich diesen Link zu den offiziellen Dokumenten für die Standardkonfigurationen.
- Lesen Sie auch diesen Link zu den offiziellen Dokumenten für die benutzerdefinierten Konfigurationen.
Dies ist ein Beispiel für eine benutzerdefinierte Konfiguration, die grundlegende Einstellungen enthält:
auto.create.topics.enable=true
default.replication.factor=3
min.insync.replicas=2
num.io.threads=8
num.network.threads=5
num.partitions=1
num.replica.fetchers=2
replica.lag.time.max.ms=30000
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true
zookeeper.session.timeout.ms=18000
Bemerkenswerte Details:
auto.create.topics.enable=true - Diese Eigenschaft hängt stark vom Arbeitsablauf und der Art und Weise ab, wie Sie Ihre Infrastruktur verwalten. Wenn Sie diese Eigenschaft nicht aktiviert haben, kann der Quellconnector das Thema nicht automatisch erstellen und Sie müssen es vorher erstellen lassen.
default.replication.factor=3 - Als Faustregel gilt, dass der Replikationsfaktor gleich der Anzahl der Broker sein sollte, um eine hohe Verfügbarkeit zu erreichen. In diesem Fall haben wir 3 Broker in 3 verschiedenen Verfügbarkeitszonen, also ist der Replikationsfaktor auf 3 gesetzt.
min.insync.replicas=2 - sollte kleiner oder gleich dem default.replication.factor
7. Quelle Anschluss
Wir werden den Debezium-Konnektor für PostgreSQL verwenden. Er wird 2 Datenbanktabellen der Quelldatenbank überwachen und alle Änderungen auf Zeilenebene erfassen. Kurz gesagt, er erstellt einen Snapshot aller Schemata und ihrer Tabellen und erfasst kontinuierlich Änderungen auf Zeilenebene, die Datenbankinhalte einfügen, aktualisieren und löschen. Dadurch wird ein Ereignisdatensatz erzeugt und als Nachricht an das Kafka-Thema gestreamt.
7.1 Benutzerdefiniertes Plugin für den Quellanschluss
Diese Ressource ist erforderlich, da sie von MSK Connect zum Importieren und Ausführen des Kafka-Connect-spezifischen Plugins, in diesem Fall Debezium, verwendet wird. Es verwendet S3 als "Backend"-Speicher, um die gezippten Kafka-Plugins zu ziehen.
Laden Sie Debezium von HIER herunter. Danach laden Sie es in Ihren S3-Bucket hoch.
7.2 Konfiguration des Quell-Connectors Worker
Worker-Konfigurationen sind eine separate AWS-Ressource im MSK-Stack der verwalteten Services. Ein Worker ist ein JVM-Prozess, der die Logik des Kafka-Connectors ausführt.
Jeder Worker kann eine Reihe von Aufgaben haben (Aufgaben.max) haben, die in parallelen Threads laufen. Die Tasks sind für das Kopieren von Daten zuständig.
Leider sind die Worker-Konfigurationen nicht sehr umfangreich, was die Möglichkeiten der Anpassung des MSK Connectors einschränkt. Eine solche Einschränkung betrifft die Konfiguration unserer Konnektoren in diesem Beispiel. Ich spreche von den internen Kafka-Themen, die das Kafka-Connect-Framework erstellt - Status, Config und Offset.
Nur das Offset-Speicher-Thema wird für die Konfiguration unterstützt, die anderen beiden werden automatisch mit eindeutigen IDs generiert, und bei der Wiederherstellung der MSK Connectors werden neue erstellt. Dies wird zu einem Problem, wenn viele Wiederherstellungen auftreten und die verfügbare Anzahl von Partitionen pro Broker, die auf dem Instanztyp des MSK-Clusters basiert, im Wesentlichen aufgefüllt wird. Kein Grund zur Panik, wenn die Anzahl der Partitionen ausgeschöpft ist. MSK Cluster wird weiterhin funktionieren, Sie können nur keine Konfigurationsänderungen vornehmen.
Automatisch generierte Status-, Konfigurations- und Offset-Topic-Namen sehen ähnlich aus wie diese:
Offset-__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
Status - __amazon_msk_connect_status_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
Config - __amazon_msk_connect_config_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
Beispiel einer Worker-Konfiguration für unseren Quell-Connector:
1 key.converter=org.apache.kafka.connect.storage.StringConverter
2 value.converter=org.apache.kafka.connect.storage.StringConverter
3 offset.storage.topic=source-datapipeline-app1-prod-offset-storage
4 offset.storage.partitions=1
5 offset.storage.replication.factor=3
Zeile 3: Wenn Sie einen Namen für den Offset-Speicher festlegen, können Sie den zuletzt übertragenen Offset beibehalten und das Lesen fortsetzen, wenn der MSK Connector neu erstellt wird.
Zeile 4: Der Standardwert ist 25, was eine Menge ist. Debezium kann nur mit "tasks.max=1" laufen, was bedeutet, dass nur 1 Partition verwendet wird. Wenn Sie die Arbeitslast verteilen wollen, müssen Sie zusätzliche Quellkonnektoren erstellen, die Änderungen aus verschiedenen Tabellen streamen.
Zeile 5: Replizieren Sie das Offset-Speicher-Thema auf allen Maklerknoten, in diesem Fall haben wir nur 3.
7.3 Konfiguration des Quellanschlusses
Kafka-Connect-Konfigurationen können je nach Anwendungsfall und Funktion drastisch variieren.
Dies ist eine Beispielkonfiguration, bei der ich versuchen werde, die Idee hinter einigen der Eigenschaften zu beschreiben:
1 "connector.class= "io.debezium.connector.postgresql.PostgresConnector"
2 "tasks.max"= "1"
3 "table.include.list"= "public.table_A,public.table_B"
4 "database.hostname"= "app_1_db.abcd77efg.us-east-1.rds.amazonaws.com"
5 "database.port"= "5432"
6 "database.user"= "<db-user"
7 "database.password"= "<db-password>"
8 "database.dbname"= "app1_db"
9 "database.server.name"= "app1"
10 "plugin.name"= "pgoutput"
11 "time.precision.mode"= "connect"
12 "slot.name"= "source_datapipeline_app_1"
13 "publication.name"= "source_datapipeline_app_1"
14 "publication.autocreate.mode"= "filtered"
15 "transforms"= "Reroute,unwrap"
16 "transforms.Reroute.type"= "io.debezium.transforms.ByLogicalTableRouter"
17 "transforms.Reroute.topic.regex"= "(.*).public.(.*)"
18 "transforms.Reroute.topic.replacement"= "$1_$2"
19 "transforms.unwrap.type"= "io.debezium.transforms.ExtractNewRecordState"
20 "transforms.unwrap.drop.tombstones"= "false"
21 "value.converter"= "io.confluent.connect.avro.AvroConverter"
22 "value.converter.basic.auth.credentials.source"= "USER_INFO"
23 "value.converter.schema.registry.url"= "https://abcd-12ef3.us-east-2.aws.confluent.cloud"
24 "value.converter.basic.auth.user.info"= "<API_KEY>:<API_SECRET>"
25 "key.converter"= "io.confluent.connect.avro.AvroConverter"
26 "key.converter.basic.auth.credentials.source"= "USER_INFO"
27 "key.converter.schema.registry.url"= "https://abcd-12ef3.us-east-2.aws.confluent.cloud"
28 "key.converter.basic.auth.user.info"= "<API_KEY>:<API_SECRET>"
Zeile 1: definiert den Typ des Kafka-Connect-Plugins, das wir verwenden werden. In unserem Fall ist es Debezium.
Zeile 2: Wie bereits erwähnt, kann Debezium nur mit einer einzigen Aufgabe arbeiten, da es zum Lesen der Änderungen aus einem einzigen pg_replication-Slot verwendet wird.
Zeile 3: Geben Sie die Datenbanktabellen an, aus denen der Connector lesen soll. Hier muss der Fully Qualified Table Name (FQTN) verwendet werden - in diesem Fall sind es die Tabellen "table_A" und "table_B" aus dem "public" Datenbankschema.
Zeile 4: Endpunkt des RDS
Zeile 5: Datenbank-Port
Zeile 6: Leider konnte ich keine Möglichkeit finden, einen Replikations- und Veröffentlichungsslot mit einem anderen Benutzer als dem ursprünglichen Master-Benutzer zu erstellen. Eine Möglichkeit besteht darin, ihn vorher zu erstellen oder das AWS Secrets Manager Plugin für AWS MSK zu verwenden und die Anmeldeinformationen dort zu speichern. Mehr dazu hier - https://aws.amazon.com/about-aws/whats-new/2022/03/amazon-msk-external-secrets-configurartion-providers/
Zeile 7: Passwort des Datenbankbenutzers
Zeile 8: Tatsächlicher Name der PostgreSQL-Datenbank, mit der Sie sich verbinden wollen
Zeile 9: Logischer Name, der einen Namespace für die bestimmte PostgreSQL-Datenbankinstanz bereitstellt und für die Bildung des Kafka-Topic-Namens verwendet wird, wenn "auto.create.topics.enable" wahr ist.
Zeile 10: Dies ist das logische Dekodierungs-Plugin. Es handelt sich um einen Mechanismus, der die Extraktion der Änderungen, die in das Transaktionsprotokoll übertragen wurden, und die Verarbeitung dieser Änderungen auf benutzerfreundliche Weise mit Hilfe eines Ausgabe-Plugins ermöglicht. Das Output-Plugin ermöglicht es den Clients, die Änderungen zu konsumieren.
In unserem Fall verwenden wir pgoutput. Dies ist das Standard-Plugin für die logische Dekodierung der Ausgabe in PostgreSQL 10+. Es wird von der PostgreSQL-Community gepflegt und von PostgreSQL selbst für die logische Replikation verwendet. Dieses Plug-in ist immer vorhanden, so dass keine zusätzlichen Bibliotheken installiert werden müssen. Der Debezium-Konnektor interpretiert den rohen Replikations-Ereignisstrom direkt in Änderungsereignisse. Die anderen Plug-ins sind decoderbufs und wal2json, auf die wir aber nicht weiter eingehen werden.
Zeile 11: Dies ist erforderlich, wenn Sie Spalten vom Typ "TIMESTAMP" mit Zeitzoneninformationen haben und das Zeitformat in der Kafka-Nachricht unverändert beibehalten wollen.
Zeile 12: Name des zu erstellenden Replikations-Slots, sofern er nicht bereits vorhanden ist. Wie man auflistet:
select * from pg_replication_slots;
Zeile 13: Name des Publikationsslots, der erstellt wird, wenn er noch nicht vorhanden ist. So wird aufgelistet:
select * from pg_publication;
select * from pg_publication_tables;
Zeile 14: empfohlen von Debezium. Der Standardwert ist "all_tables", womit eine Veröffentlichung für alle Tabellen erstellt wird. Der Wert "filtered" hingegen erstellt nur für die in der "table.include.list" aufgeführten Tabellen.
Zeilen 15-20: Einzelne Nachrichtentransformationen (SMTs), die wir hauptsächlich verwenden werden, um den Namen der automatisch erstellten Themen in Kafka mithilfe von Regex zu transformieren. Im Wesentlichen ändern wir den Namen des Topics von "app1.public.table_A" in "app1_table_A". Dies wird vom Sink Connector benötigt, da die Namen der Datenbanktabellen, die er erstellt und in die er schreibt, auf den Kafka-Topic-Namen basieren. Wenn dies nicht geändert wird, zeigt der Sink Connector den folgenden Fehler an:
org.postgresql.util.PSQLException: ERROR: datenbankübergreifende Referenzen sind nicht implementiert: "app1.public.table_A"
Zeilen 21-28: Definieren Sie Details über unser Schema-Register, wie URL, Anmeldeinformationen und Art der Authentifizierung. Wir verwenden die einfache Authentifizierung und setzen sie daher auf "USER_INFO".
Aufgetretene Fehler und ihre Lösungen:
“ERROR Postgres roles LOGIN and REPLICATION are not assigned to the user”:
Lösung: Diese Art von Fehler kann auftreten, wenn der Quellconnector versucht, neue Replikations- und Veröffentlichungsslots zu erstellen. Eine Möglichkeit, dieses Problem zu lösen, besteht darin, die Slots vorher manuell oder mit einer Art Automatisierung (Ansible/Puppet/Chef) zu erstellen. Die zweite Möglichkeit besteht darin, die Erstellung der Veröffentlichungs- und Replikationsslots dem Connector zu überlassen und entweder einen dedizierten Benutzer mit "REPLICATION"-Berechtigungen (folgen Sie dieser Anleitung https://debezium.io/documentation/reference/stable/connectors/postgresql.html#postgresql-permissions) oder den ursprünglichen Master-Benutzer des RDS zu verwenden.
ERROR Postgres server wal_level property must be "logical" but is: replica (io.debezium.connector.postgresql.PostgresConnector:101)
Lösung: Die logische Replikation muss aktiviert werden. Dies kann nur durch die Anwendung einer Parametergruppe (einzelne RDS-Instanz) oder einer DB-Cluster-Parametergruppe (Aurora-Cluster) und die Einstellung "rds.logical_replication=1" erfolgen. Danach muss RDS neu gebootet werden, damit die Änderung wirksam wird.
7.4 Sink Connector Worker Konfiguration
Die Worker-Konfiguration für den Sink-Connector unterscheidet sich nicht von der für den Source-Connector, mit Ausnahme der Eigenschaft "offset.storage.partitions".
Beispiel:
1 key.converter=org.apache.kafka.connect.storage.StringConverter
2 value.converter=org.apache.kafka.connect.storage.StringConverter
3 offset.storage.topic=sink-datapipeline-app1-prod-offset-storage
4 offset.storage.partitions=1
5 offset.storage.replication.factor=3
Line 3: Setting a name for the offset storage allows you to keep the last committed offset and continue reading if a recreation of the MSK Connector occurs.
Line 4: Unlike Debezium, the JDBC sink connector can run with more than “tasks.max=1”, because all tasks working in the connector are in a single consumer group, giving you the opportunity to achieve higher throughput by a single sink connector, instead of creating multiple connectors for the different Kafka topics. The consumer group’s ID is formed by the name of the MSK connector prepended by “connect-”. In our case, it will be “connect-sink-datapipeline-app1-prod”. If you plan to use more than one “tasks.max”, you need to increase the number of partitions for the offset storage topic, so that “tasks.max” <= “offset.storage.partitions”.
Zeile 5: Replizieren Sie das Offset-Speicher-Thema auf allen Maklerknoten, in diesem Fall haben wir nur 3.
7.5 Konfiguration des Senkenanschlusses
Für den Sink-Connector, der verwendet wird, um die Nachrichten aus dem Kafka-Thema zu konsumieren und sie in den PostgreSQL-RDS zu schreiben, werden wir das JDBC-Sink-Connector-Plugin von Confluent verwenden. Genau wie bei der Konfiguration des Quellkonnektors werde ich versuchen, eine detailliertere Sicht auf jede Kafka-Verbindungseigenschaft zu geben und später einige der aufgetretenen Probleme und ihre Lösungen aufzulisten.
Kafka verbindet Plugin herunterladen HIER. Danach müssen Sie die Zip-Datei in Ihren S3-Bucket hochladen und eine AWS Custom Plugin-Ressource erstellen, die vom Sink Connector verwendet wird.
Senke Anschlusskonfiguration:
1 "connector.class"= "io.confluent.connect.jdbc.JdbcSinkConnector"
2 "tasks.max"= "1"
3 "topics"= "app1_table_A,app1_table_B"
4 "table.name.format"= "${topic}"
5 "connection.url"= "jdbc:postgresql://target-db.abc1defghijk.us-east-1.rds.amazonaws.com:5432/target_db"
6 "connection.user"= "<db-user>"
7 "connection.password"= "<db-user-password>"
8 "delete.enabled"= "true"
9 "auto.create"= "true"
10 "auto.evolve"= "true"
11 "insert.mode"= "upsert"
12 "pk.mode"= "record_key"
13 "pk.fields"= "uuid"
14 "value.converter.basic.auth.credentials.source"= "USER_INFO"
15 "value.converter.schema.registry.url"= "https://abcd-12ef3.us-east-2.aws.confluent.cloud"
16 "value.converter.basic.auth.user.info"= "<API_KEY>:<API_SECRET>"
17 "key.converter"= "io.confluent.connect.avro.AvroConverter"
18 "key.converter.basic.auth.credentials.source"= "USER_INFO"
19 "key.converter.schema.registry.url"= "https://abcd-12ef3.us-east-2.aws.confluent.cloud"
20 "key.converter.basic.auth.user.info"= "<API_KEY>:<API_SECRET>"
Line 1: define the connector class. In this case, it is JDBC Sink Connector
Line 2: as mentioned earlier sink connectors can work with more than one task, but they need to be equal to or less than the offset.storage.partitions settings in the worker configuration. If they are more the extra tasks will be idle as there are no available partitions to connect to.
Line 3: List the Kafka topics that the connector will consume from
Line 4: Default value is “${topic}”. This is a format string for the destination table name. This works alongside the “topics” property and will create a separate database table for each topic listed in it. If you are using terraform to deploy the connector you need to escape the value like this – “$${topic}”.
Line 5: In short, this is the endpoint of the database, and “target_db” is the target database that we want to write to.
Line 6: Database user
Line 7: Database user’s password
Line 8: (Depends on use-case) enable deletion of records.
Line 9: enables the auto-creation of the destination table if it’s missing. If you disable this property you need to have the destination tables created beforehand, either manually or by some automation such as Ansible, Chef, Puppet, Saltstack, and so on.
Line 10: from the official documentation – “the connector can perform limited auto-evolution by issuing ALTER on the destination table when it encounters a record for which a column is found to be missing. Since data-type changes and the removal of columns can be dangerous, the connector does not attempt to perform such evolutions on the table. The addition of primary key constraints is also not attempted. In contrast, if “auto.evolve” is disabled no evolution is performed and the connector task fails with an error stating the missing columns.”
Line 11: from the official documentation – “Upsert semantics refer to atomically adding a new row or updating the existing row if there is a primary key constraint violation, which provides idempotence.”
Lines 12 and 13: Please, refer to official documentation, as it is best described there – https://docs.confluent.io/kafka-connectors/jdbc/current/sink-connector/overview.html#delete-mode
Lines 14 to 20: define details about our Schema Registry, such as URL, credentials, and type of authentication. We will be using basic auth, so we are setting it to “USER_INFO”.
Aufgetretene Fehler und ihre Lösungen:
1. Caused by: java.lang.ClassNotFoundException: com.google.common.base.Ticker
Lösung: Vergewissern Sie sich, dass Sie ein "Guava"-Glas in Ihrem Spülenanschluss-Zip-Bündel haben.
2. Failed to send HTTP request to endpoint
Lösung: Dieser Fehler kann sowohl beim Sink- als auch beim Source-Connector auftreten und ist wahrscheinlich darauf zurückzuführen, dass der Connector in einem öffentlichen Subnetz mit einem Internet-Gateway eingesetzt wird. Sie müssen ihn in einem privaten Subnetz hinter einem NAT-Gateway bereitstellen, da er sonst keine Internetverbindung hat.
8. Überwachung
Obwohl AWS MSK mit Tools wie AWS Cloudwatch und Prometheus überwacht werden kann, möchte ich Ihnen eine praktikable Alternative vorstellen - AKHQ.
Schauen Sie sich die offizielle Website hier und Github an.
AKHQ ist Open-Source und bietet eine einfache und intuitive Schnittstelle für die Verwaltung von Kafka-Clustern und vereinfacht viele der Aufgaben, die normalerweise über die Kafka-Befehlszeilenschnittstelle (CLI) ausgeführt werden. Sie ermöglicht die Überwachung von Themen, die Anzeige von Verbrauchergruppen, die Überprüfung von Nachrichten und die Durchführung verschiedener administrativer Aufgaben, wie die Erstellung neuer Themen oder das Löschen bestehender Themen.
Einige der wichtigsten Vorteile der Verwendung von AKHQ für die Verwaltung von Kafka-Clustern sind:
- Benutzerfreundliche Schnittstelle: AKHQ bietet eine benutzerfreundliche Oberfläche, die die Verwaltung von Kafka-Clustern vereinfacht und es Entwicklern, Administratoren und anderen Benutzern erleichtert, Kafka-Themen und Verbrauchergruppen zu verwalten und zu überwachen.
- Zentralisierte Verwaltung: Mit AKHQ können Sie mehrere Kafka-Cluster von einer einzigen webbasierten Oberfläche aus verwalten, was die Verwaltung und Überwachung umfangreicher Implementierungen erleichtert.
- Überwachung in Echtzeit: AKHQ bietet Echtzeit-Überwachung von Kafka-Themen und Verbrauchergruppen, so dass Sie Probleme schnell erkennen und beheben können, sobald sie auftreten.
- Erweiterbarkeit: AKHQ ist in hohem Maße erweiterbar und kann an die spezifischen Anforderungen Ihres Unternehmens angepasst werden.
Beispiel für Steuerwerte:
image:
repository: tchiotludo/akhq
configuration:
micronaut:
security:
enabled: true
token:
jwt:
signatures:
secret:
generator:
secret: "<secret>" #Changeme
akhq:
security:
enabled: true
default-group: no-roles
basic-auth:
- username: admin
password: <sha256sum> #Changeme
groups:
- admin
- username: readonly
password: <sha256sum> #Changeme
groups:
- reader
pagination:
page-size: 10
threads: 16
clients-defaults:
consumer:
properties:
default.api.timeout.ms: 60000
max.poll.records: 25
server:
access-log:
enabled: true
name: org.akhq.log.access
existingSecrets: ""
secrets:
akhq:
connections:
kafka-cluster:
properties:
bootstrap.servers: "b-1.kafkacluster.abcdef.a11.kafka.us-east-1.amazonaws.com:9092,b-2.kafkacluster.abcdef.a11.kafka.us-east-1.amazonaws.com:9092,b-3.kafkacluster.abcdef.a11.kafka.us-east-1.amazonaws.com:9092"
request.timeout.ms: 60000
schema-registry:
url: "https://abcd-12ef3.us-east-2.aws.confluent.cloud"
type: "confluent"
basic-auth-username: <API_KEY>
basic-auth-password: <API_SECRET>
service:
enabled: true
type: ClusterIP
port: 80
labels: {}
annotations:
ingress:
enabled: true
ingressClassName: ""
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
kubernetes.io/ingress.class: nginx
kubernetes.io/tls-acme: "true"
paths:
- /
hosts:
- my.domain.com #Changeme
tls:
- secretName: my-domain-com-tls #Changeme
hosts:
- my.domain.com #Changeme
readinessProbe:
enabled: true
path: /health
port: management
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
successThreshold: 1
failureThreshold: 3
httpGetExtra: {}
livenessProbe:
enabled: true
path: /health
port: management
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
successThreshold: 1
failureThreshold: 3
httpGetExtra: {}
resources:
limits:
cpu: 1000m
memory: 1024Mi
requests:
cpu: 500m
memory: 512Mi
networkPolicy:
enabled: true
Weitere Optionen von GitHub können Sie hier entdecken.
Schlussfolgerung
Sie können Kafka-basierte Datenpipelines in der Cloud mit Hilfe von AWS MSK und AWS MSK Connect, beides leistungsstarke Lösungen, erstellen und verwalten. Skalierbarkeit, hohe Verfügbarkeit, optimierte Verwaltung, Konnektivität mit anderen Amazon-Services und benutzerfreundliche Konnektoren sind nur einige der Vorteile, die sie bieten. Die Nachteile sind die Kosten, die Lernkurve, die Bindung an einen bestimmten Anbieter und die mangelnde Anpassungsfähigkeit. Bei der Entscheidung, ob Amazon MSK und AWS MSK Connect die besten Optionen für die Datenverarbeitungsanforderungen Ihres Unternehmens sind, ist es wichtig, die Vor- und Nachteile der Integration sorgfältig abzuwägen.
Wenn Sie Unterstützung oder Beratung bei der Implementierung dieser speziellen Lösung benötigen, wenden Sie sich an unser Team erfahrener DevOps-Experten.
Ich empfehle dringend, die Blogs von Robin Moffattzu lesen, da sie mir unschätzbare Einblicke und Kenntnisse über Kafka und Kafka Connect vermittelt haben.
Eine Antwort
Regain Funds LLC verfügt über eine nachgewiesene Erfolgsbilanz bei der erfolgreichen Beilegung finanzieller Streitigkeiten und der Wiedererlangung von Vermögenswerten durch eine Kombination aus rechtlichen und technischen Mitteln.