Hallo, Habr! Heute werden wir ein System aufbauen, das mit Spark Streaming Nachrichtenströme von Apache Kafka verarbeitet und die Ergebnisse in einer Cloud-Datenbank von AWS RDS speichert.
Stellen wir uns vor, eine Kreditorganisation hat die Aufgabe, eingehende Transaktionen "in Echtzeit" in all ihren Filialen zu verarbeiten. Dies kann zum Zwecke der zeitnahen Berechnung offener Währungspositionen für das Treasury, Limiten oder finanzieller Ergebnisse aus Transaktionen und so weiter erfolgen.
Wie wir diesen Anwendungsfall ohne Magie und Zaubersprüche implementieren können – lesen Sie weiter unten! Los geht's!
Einführung
Die Verarbeitung großer Datenmengen in Echtzeit bietet zweifellos zahlreiche Möglichkeiten für moderne Systeme. Eine der beliebtesten Kombinationen dafür ist das Duo aus Apache Kafka und Spark Streaming, wobei Kafka einen Strom von eingehenden Nachrichtenpaketen erstellt und Spark Streaming diese Pakete in festgelegten Zeitintervallen verarbeitet.
Um die Fehlertoleranz der Anwendung zu erhöhen, verwenden wir Kontrollpunkte — Checkpoints. Mit diesem Mechanismus muss das Spark Streaming-Modul, wenn es verlorene Daten wiederherstellen muss, nur zu dem letzten Kontrollpunkt zurückkehren und die Berechnungen von dort aus fortsetzen.
Architektur des zu entwickelnden Systems
Verwendete Komponenten:
- — ist ein verteiltes Nachrichtenaustauschsystem mit Publish-Subscribe-Funktionalität. Es eignet sich sowohl für den Offline- als auch für den Online-Nachrichtenverbrauch. Um Datenverlust zu vermeiden, werden Kafka-Nachrichten auf der Festplatte gespeichert und innerhalb des Clusters repliziert. Das Kafka-System basiert auf dem Synchronisationsdienst ZooKeeper;
- — Komponente Spark zur Verarbeitung von Streaming-Daten. Das Spark Streaming-Modul basiert auf einer „Micro-Batch“-Architektur, bei der der Datenstrom als kontinuierliche Sequenz kleiner Datenpakete interpretiert wird. Spark Streaming empfängt Daten aus verschiedenen Quellen und vereint sie in kleinen Paketen. Neue Pakete werden in regelmäßigen Abständen erstellt. Zu Beginn jedes Zeitintervalls wird ein neues Paket erstellt, und alle während dieses Intervalls eingehenden Daten werden in dieses Paket aufgenommen. Am Ende des Intervalls wird die Paketvergrößerung gestoppt. Die Dauer des Intervalls wird durch einen Parameter bestimmt, der als Batch-Intervall bezeichnet wird;
- — kombiniert relationale Verarbeitung mit funktionalem Programmieren in Spark. Unter strukturierten Daten versteht man Daten mit einem Schema, d.h. einem einheitlichen Satz von Feldern für alle Datensätze. Spark SQL unterstützt den Import aus vielen Quellen strukturierten Daten und kann aufgrund der Vorhandensein von Schema-Informationen effizient nur die benötigten Felder der Datensätze extrahieren und bietet auch APIs für DataFrames an;
- ist eine vergleichsweise kostengünstige Cloud-relational Datenbank, ein Webdienst, der die Konfiguration, den Betrieb und die Skalierung vereinfacht und direkt von Amazon verwaltet wird.
Einrichten und Starten des Kafka-Servers
Bevor Sie Kafka direkt verwenden, stellen Sie sicher, dass Java vorhanden ist, da die JVM verwendet wird:
sudo apt-get update
sudo apt-get install default-jre
java -version
Erstellen wir einen neuen Benutzer, um mit Kafka zu arbeiten:
sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo
Laden Sie als nächstes das Paket von der offiziellen Apache Kafka-Website herunter:
wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"Entpacken Sie das heruntergeladene Archiv:
tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka
Der nächste Schritt ist optional. Die Standardkonfiguration ermöglicht es nicht, alle Funktionen von Apache Kafka vollständig zu nutzen. Beispielsweise können Themen, Kategorien und Gruppen, auf die Nachrichten veröffentlicht werden können, nicht gelöscht werden. Um dies zu ändern, bearbeiten wir die Konfigurationsdatei:
vim ~/kafka/config/server.propertiesFügen Sie am Ende der Datei Folgendes hinzu:
delete.topic.enable = trueBevor Sie den Kafka-Server starten, müssen Sie den ZooKeeper-Server starten. Wir verwenden dazu ein Hilfsskript, das zusammen mit der Kafka-Distribution geliefert wird:
Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties
Nachdem ZooKeeper erfolgreich gestartet wurde, starten wir den Kafka-Server in einem separaten Terminal:
bin/kafka-server-start.sh config/server.propertiesLassen Sie uns ein neues Thema mit dem Namen Transaction erstellen:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transactionStellen Sie sicher, dass das Thema mit der entsprechenden Anzahl an Partitionen und Replikationen erstellt wurde:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 
Wir überspringen die Testphasen für den Producer und Consumer des neu erstellten Themas. Details dazu, wie Sie das Senden und Empfangen von Nachrichten testen können, finden Sie in der offiziellen Dokumentation — . Jetzt gehen wir zum Schreiben des Producers in Python unter Verwendung des KafkaProducer-APIs über.
Schreiben eines Producers
Der Producer wird zufällige Daten generieren — jeweils 100 Nachrichten pro Sekunde. Unter zufälligen Daten verstehen wir ein Dictionary, das aus drei Feldern besteht:
- Zweigstelle — Bezeichnung der Verkaufsstelle der Kreditinstitution;
- Währung — Währung des Geschäfts;
- Betrag — Transaktionsbetrag. Der Betrag ist eine positive Zahl, wenn die Währung vom Bank gekauft wird, und negativ, wenn sie verkauft wird.
Der Code für den Produzenten sieht folgendermaßen aus:
from numpy.random import choice, randint
def get_random_value():
new_dict = {}
branch_list = ['Kazan', 'SPB', 'Novosibirsk', 'Surgut']
currency_list = ['RUB', 'USD', 'EUR', 'GBP']
new_dict['branch'] = choice(branch_list)
new_dict['currency'] = choice(currency_list)
new_dict['amount'] = randint(-100, 100)
return new_dict
Dann verwenden wir die Methode send, um eine Nachricht im JSON-Format an den Server zu senden, in das gewünschte Thema:
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
value_serializer=lambda x:dumps(x).encode('utf-8'),
compression_type='gzip')
my_topic = 'transaction'
data = get_random_value()
try:
future = producer.send(topic = my_topic, value = data)
record_metadata = future.get(timeout=10)
print('--> Die Nachricht wurde an ein Thema gesendet:
{}, Partition: {}, Offset: {}'
.format(record_metadata.topic,
record_metadata.partition,
record_metadata.offset ))
except Exception as e:
print('--> Es scheint ein Fehler aufgetreten zu sein: {}'.format(e))
finally:
producer.flush()
Wenn das Skript ausgeführt wird, erhalten wir die folgenden Nachrichten im Terminal:
Das bedeutet, dass alles wie gewünscht funktioniert — der Produzent generiert und sendet Nachrichten an das gewünschte Thema.
Der nächste Schritt besteht in der Installation von Spark und der Verarbeitung dieses Nachrichtenstroms.
Installation von Apache Spark
Apache Spark ist eine universelle und leistungsstarke Cluster-Computing-Plattform.
In Bezug auf die Leistung übertrifft Spark gängige Implementierungen des MapReduce-Modells und ermöglicht gleichzeitig die Unterstützung einer breiteren Palette von Berechnungsarten, einschließlich interaktiver Anfragen und Streaming-Verarbeitung. Geschwindigkeit spielt eine entscheidende Rolle bei der Verarbeitung großer Datenmengen, da sie interaktives Arbeiten ermöglicht, ohne Minuten oder Stunden auf das Ergebnis warten zu müssen. Eines der wichtigsten Merkmale von Spark, das eine so hohe Geschwindigkeit gewährleistet, ist die Fähigkeit zur Durchführung von Berechnungen im Arbeitsspeicher.
Dieses Framework ist in Scala geschrieben, daher muss diese zuerst installiert werden:
sudo apt-get install scalaLaden Sie das Spark-Distributionspaket von der offiziellen Website herunter:
wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"Entpacken Sie das Archiv:
sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/sparkFügen Sie den Pfad zu Spark in die Bash-Datei ein:
vim ~/ .bashrcFügen Sie über den Editor die folgenden Zeilen hinzu:
SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH
Führen Sie den folgenden Befehl aus, nachdem Sie Änderungen in der bashrc vorgenommen haben:
source ~/.bashrcBereitstellung von AWS PostgreSQL
Nun müssen wir die Datenbank bereitstellen, in die wir die verarbeiteten Informationen aus den Streams einpflegen werden. Dafür verwenden wir den AWS RDS-Dienst.
Gehen Sie zur AWS-Konsole -> AWS RDS -> Datenbanken -> Datenbank erstellen:
Wählen Sie PostgreSQL und klicken Sie auf die Schaltfläche Weiter:
Da dieses Beispiel ausschließlich zu Bildungszwecken dient, verwenden wir einen kostenlosen „Basis“-Server (Free Tier):
Setzen Sie das Häkchen im Bereich Free Tier, und Ihnen wird automatisch eine Instanz der Klasse t2.micro angeboten – zwar schwach, aber kostenlos und somit für unsere Aufgabe geeignet:
Als Nächstes sind sehr wichtige Dinge erforderlich: der Name der DB-Instanz, der Name des Master-Benutzers und dessen Passwort. Wir nennen die Instanz: myHabrTest, Master-Benutzer: habr, Passwort: habr12345 und klicken auf die Schaltfläche Weiter:
Auf der nächsten Seite finden Sie die Einstellungen für die externe Verfügbarkeit unseres DB-Servers (Öffentliche Erreichbarkeit) und die Erreichbarkeit von Ports:
Lassen Sie uns eine neue Einstellung für die VPC-Sicherheitsgruppe erstellen, die es ermöglicht, von außen über Port 5432 (PostgreSQL) auf unseren DB-Server zuzugreifen.
Öffnen Sie in einem neuen Browserfenster die AWS-Konsole und gehen Sie zu VPC-Dashboard —> Sicherheitsgruppen —> Sicherheitsgruppe erstellen:
Geben Sie einen Namen für die Sicherheitsgruppe — PostgreSQL, eine Beschreibung ein, wählen Sie die zugehörige VPC aus und klicken Sie auf die Schaltfläche Erstellen:
Füllen Sie für die neu erstellte Gruppe die eingehenden Regeln für Port 5432 aus, wie im Bild unten gezeigt. Sie können den Port manuell nicht angeben und PostgreSQL aus dem Dropdown-Menü Typ auswählen.
Streng genommen bedeutet der Wert ::/0, dass eingehender Verkehr für den Server von überall auf der Welt verfügbar ist, was kanonisch nicht ganz korrekt ist, aber für illustrative Zwecke erlauben wir uns diesen Ansatz:
Gehen Sie zurück zur Browser-Seite, wo wir „Erweiterte Einstellungen konfigurieren“ geöffnet haben, und wählen Sie im Abschnitt VPC-Sicherheitsgruppen —> Vorhandene VPC-Sicherheitsgruppen auswählen —> PostgreSQL:
Weiterhin, im Abschnitt Datenbankoptionen —> Datenbankname —> geben Sie den Namen ein — habrDB.
Die übrigen Parameter, mit Ausnahme der Deaktivierung der Backup-Speicherung (Backupaufbewahrungszeitraum — 0 Tage), der Überwachung und der Performance Insights, können auf den Standardwerten belassen werden. Klicken Sie auf die Schaltfläche Datenbank erstellen:
Stream-Handler
Der letzte Schritt besteht darin, einen Spark-Job zu entwickeln, der alle zwei Sekunden neue Daten von Kafka verarbeitet und die Ergebnisse in die Datenbank einfügt.
Wie bereits erwähnt, sind Checkpoints die grundlegende Mechanismus bei Spark Streaming, die eingerichtet werden müssen, um die Ausfallsicherheit zu gewährleisten. Wir werden Checkpoints verwenden, und im Falle eines Fehlers muss man nur zum letzten Checkpoint zurückkehren und die Berechnungen von dort aus fortsetzen.
Ein Checkpoint kann aktiviert werden, indem ein Verzeichnis in einem ausfallsicheren, zuverlässigen Dateisystem (z. B. HDFS, S3 usw.) festgelegt wird, in dem die Informationen des Checkpoints gespeichert werden. Das wird beispielsweise so gemacht:
streamingContext.checkpoint(checkpointDirectory)In unserem Beispiel verwenden wir den folgenden Ansatz: Wenn das checkpointDirectory existiert, wird der Kontext aus den Daten des Checkpoints wiederhergestellt. Wenn das Verzeichnis nicht existiert (d.h. es wird zum ersten Mal ausgeführt), wird die Funktion functionToCreateContext aufgerufen, um einen neuen Kontext zu erstellen und die DStreams zu konfigurieren.
from pyspark.streaming import StreamingContext
context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)
Wir erstellen ein DirectStream-Objekt, um sich über die Methode createDirectStream der KafkaUtils-Bibliothek mit dem Topic „transaction“ zu verbinden:
from pyspark.streaming.kafka import KafkaUtils
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 2)
broker_list = 'localhost:9092'
topic = 'transaction'
directKafkaStream = KafkaUtils.createDirectStream(ssc,
[topic],
{"metadata.broker.list": broker_list})
Wir parsen die eingehenden Daten im JSON-Format:
rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
currency=w['currency'],
amount=w['amount']))
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")
Mit Spark SQL führen wir eine einfache Gruppierung durch und geben das Ergebnis in der Konsole aus:
select
from_unixtime(unix_timestamp()) as curr_time,
t.branch as branch_name,
t.currency as currency_code,
sum(amount) as batch_value
from treasury_stream t
group by
t.branch,
t.currency
Wir erhalten den Text der Anfrage und führen ihn über Spark SQL aus:
sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)
Anschließend speichern wir die aggregierten Daten in einer Tabelle in AWS RDS. Um die Aggregationsergebnisse in einer Datenbanktabelle zu speichern, verwenden wir die Methode write des DataFrame-Objekts:
testResultDataFrame.write
.format("jdbc")
.mode("append")
.option("driver", 'org.postgresql.Driver')
.option("url","jdbc:postgresql://myhabrtest.ciny8bykwxeg.us-east-1.rds.amazonaws.com:5432/habrDB")
.option("dbtable", "transaction_flow")
.option("user", "habr")
.option("password", "habr12345")
.save()
Einige Worte zur Konfiguration der Verbindung zu AWS RDS. Benutzername und Passwort wurden in der Phase „Bereitstellung von AWS PostgreSQL“ erstellt. Als URL für den Datenbankserver sollte der Endpoint verwendet werden, der im Abschnitt Connectivity & security angezeigt wird:
Um eine korrekte Verbindung zwischen Spark und Kafka herzustellen, sollte der Job über spark-submit mit dem Artefakt gestartet werden spark-streaming-kafka-0-8_2.11. Zusätzlich verwenden wir ein Artefakt zur Interaktion mit der PostgreSQL-Datenbank, das wir über —packages übergeben werden.
Für die Flexibilität des Skripts werden wir auch den Namen des Nachrichtendienstes und das Thema, aus dem wir Daten abrufen möchten, als Eingabeparameter auslagern.
So, es ist an der Zeit, das System zu starten und die Funktionsfähigkeit zu überprüfen:
spark-submit
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2,
org.postgresql:postgresql:9.4.1207
spark_job.py localhost:9092 transaction
Alles hat funktioniert! Wie auf dem Bild unten zu sehen ist, werden während der Nutzung der Anwendung alle 2 Sekunden neue Aggregationsergebnisse angezeigt, da wir beim Erstellen des StreamingContext ein Paketintervall von 2 Sekunden festgelegt haben.
Als Nächstes führen wir eine einfache Abfrage an der Datenbank durch, um das Vorhandensein von Datensätzen in der Tabelle zu überprüfen. transaction_flow:
Fazit
In diesem Artikel wurde ein Beispiel für die Streamverarbeitung von Informationen mit Spark Streaming in Kombination mit Apache Kafka und PostgreSQL behandelt. Angesichts des Wachstums der Datenmengen aus verschiedenen Quellen ist der praktische Wert von Spark Streaming für die Entwicklung von Streaming-Anwendungen und Anwendungen in Echtzeit kaum zu überschätzen.
Den vollständigen Quellcode finden Sie in meinem Repository auf .
Ich freue mich, diesen Artikel zu besprechen und warte auf Ihre Kommentare. Zudem hoffe ich auf konstruktive Kritik von allen interessierten Lesern.
Viel Erfolg!
Ps. Ursprünglich war geplant, eine lokale PostgreSQL-Datenbank zu verwenden, aber aufgrund meiner Vorliebe für AWS habe ich mich entschieden, die Datenbank in die Cloud auszulagern. In dem nächsten Artikel zu diesem Thema zeige ich, wie man das oben beschriebene System vollständig in AWS mit AWS Kinesis und AWS EMR implementiert. Bleiben Sie dran!
Quelle: habr.com
