Ahoj, Habr! Dnes postavíme systém, který pomocí Spark Streaming zpracovává streamy zpráv Apache Kafka a ukládá výsledky zpracování do cloudové databáze AWS RDS.
Představme si, že nás nějaká kreditní instituce požádá o zpracování příchozích transakcí „za chodu“ ve všech svých pobočkách. To může být provedeno za účelem okamžitého výpočtu otevřené měnové pozice pro pokladnu, limitů nebo finančního výsledku transakcí atd.
Jak tento případ realizovat bez použití magie a kouzelných zaklínadel — čtěte pod katem! Jdeme na to!
Úvod
Jednoznačně, zpracování velkých objemů dat v reálném čase poskytuje široké možnosti pro využití v moderních systémech. Jednou z nejoblíbenějších kombinací pro toto je tandem Apache Kafka a Spark Streaming, kde Kafka vytváří stream balíků příchozích zpráv, zatímco Spark Streaming zpracovává tyto balíky v daném časovém intervalu.
Abychom zvýšili odolnost aplikace, budeme používat kontrolní body — checkpointy. Pomocí tohoto mechanismu, když modulu Spark Streaming bude potřeba obnovit ztracená data, bude se muset pouze vrátit k poslednímu kontrolnímu bodu a obnovit výpočty od něj.
Architektura vyvíjeného systému
Používané komponenty:
- — je to distribuovaný systém výměny zpráv s publikováním a odběrem. Je vhodný jak pro offline, tak pro online spotřebu zpráv. Aby se zabránilo ztrátě dat, zprávy Kafka se ukládají na disk a replikují uvnitř klastru. Systém Kafka je postaven na službě synchronizace ZooKeeper;
- — komponent Spark pro zpracování streamovaných dat. Modul Spark Streaming je postaven na „mikrobalíčkové“ architektuře, kdy je stream dat interpretován jako kontinuální sekvence malých balíčků dat. Spark Streaming přijímá data z různých zdrojů a kombinuje je do malých balíčků. Nové balíčky se vytvářejí v pravidelných časových intervalech. Na začátku každého časového intervalu se vytvoří nový balíček a jakákoliv data přijatá během tohoto intervalu jsou zahrnuta do balíčku. Na konci intervalu se zvyšování balíčku zastaví. Velikost intervalu je určena parametrem, který se nazývá interval balíčkování;
- — spojuje relační zpracování s funkcionálním programováním Spark. Pod strukturovanými daty se rozumí data, která mají schéma, tedy jednotnou sadu polí pro všechny záznamy. Spark SQL podporuje vstup z mnoha zdrojů strukturovaných dat a díky informacím o schématu může efektivně extrahovat pouze potřebná pole záznamů a také poskytuje rozhraní API DataFrame;
- — je relativně levá cloudová relační databáze, webová služba, která zjednodušuje nastavení, provoz a škálování, spravována přímo Amazonem.
Nastavení a spuštění serveru Kafka
Před přímým použitím Kafka je nutné se ujistit v přítomnosti Javy, protože pro provoz se používá JVM:
sudo apt-get update
sudo apt-get install default-jre
java -version
Vytvoříme nového uživatele pro práci s Kafkou:
sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo
Dále stáhneme distribuci z oficiálních stránek Apache Kafka:
wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"Rozbalíme stažený archiv:
tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka
Další krok je volitelný. Věc je taková, že výchozí nastavení neumožňují plně využít všech možností Apache Kafka. Například odstraňovat téma, kategorii, skupinu, na které mohou být publikovány zprávy. Abychom to změnili, upravíme konfigurační soubor:
vim ~/kafka/config/server.propertiesPřidejte na konec souboru následující:
delete.topic.enable = truePřed spuštěním serveru Kafka je třeba spustit server ZooKeeper. Použijeme pomocný skript, který je dodáván s distrubucí Kafka:
Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties
Jakmile ZooKeeper úspěšně běží, spustíme server Kafka v samostatném terminálu:
bin/kafka-server-start.sh config/server.propertiesVytvoříme nový topic s názvem Transaction:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transactionUjistíme se, že topic byl vytvořen s požadovaným počtem partition a replikací:
bin/kafka-topics.sh --describe --zookeeper localhost:2181 
Vynecháme okamžiky testování producenta a konzumenta pro nově vytvořený topic. Podrobnosti o tom, jak testovat odesílání a příjem zpráv, jsou uvedeny v oficiální dokumentaci — . A nyní přejdeme k psaní producenta v Pythonu pomocí API KafkaProducer.
Psaní producenta
Producent bude generovat náhodná data — 100 zpráv každou sekundu. Náhodnými daty budeme mít na mysli slovník skládající se ze tří polí:
- Pobočka — název prodejního místa úvěrové organizace;
- Currency — měna transakce;
- Amount — částka transakce. Částka bude kladné číslo, pokud jde o nákup měny bankou, a záporné číslo, pokud se jedná o prodej.
Kód pro producenta vypadá takto:
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
Dále, pomocí metody send, odešleme zprávu na server do požadovaného topicu ve formátu JSON:
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('--> Zpráva byla odeslána do topicu:
{}, partition: {}, offset: {}'
.format(record_metadata.topic,
record_metadata.partition,
record_metadata.offset ))
except Exception as e:
print('--> Zda se, že došlo k chybě: {}'.format(e))
finally:
producer.flush()
Při spuštění skriptu získáme v terminálu následující zprávy:
To znamená, že vše funguje, jak jsme chtěli — producent generuje a odesílá zprávy do požadovaného topicu.
Dalším krokem bude instalace Sparku a zpracování tohoto proudu zpráv.
Instalace Apache Spark
Apache Spark — je univerzální a vysoce výkonná platforma pro výpočty v clusteru.
Co se výkonu týče, Spark překonává populární implementace modelu MapReduce a zároveň podporuje širší škálu typů výpočtů, včetně interaktivních dotazů a streamového zpracování. Rychlost hraje důležitou roli při zpracování velkých objemů dat, protože právě rychlost umožňuje pracovat interaktivně, aniž bychom museli čekat minuty nebo hodiny. Jednou z nejdůležitějších předností Sparku, která zajišťuje tak vysokou rychlost, je schopnost provádět výpočty v paměti.
Tento framework je napsán ve Scalě, a proto je nutné začít instalací Scala:
sudo apt-get install scalaStáhněte si z oficiálních stránek distribuci Sparku:
wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"Rozbalte archiv:
sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/sparkPřidejte cestu k Sparku do bash souboru:
vim ~/ .bashrcPřidejte následující řádky přes editor:
SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH
Po provedení úprav v bashrc spusťte níže uvedený příkaz:
source ~/ .bashrcNasazení AWS PostgreSQL
Zbývá nasadit databázi, do které budeme ukládat zpracované informace z toků. K tomu použijeme službu AWS RDS.
Přejděte do konzole AWS —> AWS RDS —> Databases —> Vytvořit databázi:
Vyberte PostgreSQL a klikněte na tlačítko Další:
Protože tento příklad je zaměřen výhradně na vzdělávací účely, použijeme zdarma server 'na minimálkách' (Free Tier):
Následně zaškrtněte políčko v bloku Free Tier a poté nám automaticky bude nabízen instanc třídy t2.micro — i když slabší, je zdarma a naprosto vyhovující pro náš úkol:
Následují velmi důležité informace: název instancí DB, jméno správce a jeho heslo. Nazveme instanci: myHabrTest, správce: habr, heslo: habr12345 a klikněte na tlačítko Další:
Na další stránce jsou parametry, které zajišťují dostupnost našeho databázového serveru zvenčí (Veřejná dostupnost) a dostupnost portů:
Pojďme vytvořit nové nastavení pro VPC security group, které umožní přístup k našemu databázovému serveru zvenčí přes port 5432 (PostgreSQL).
Otevřete v samostatném okně prohlížeče konzoli AWS, přejděte na VPC Dashboard —> Skupiny zabezpečení —> Vytvořit skupinu zabezpečení:
Nastavujeme název pro skupinu zabezpečení — PostgreSQL, popis, uvádíme, ke které VPC by tato skupina měla být přidružena, a klikneme na tlačítko Vytvořit:
Vyplňujeme pro nově vytvořenou skupinu Příchozí pravidla pro port 5432, jak je znázorněno na obrázku níže. Port není nutné zadávat ručně, můžete vybrat PostgreSQL z rozbalovacího seznamu Typ.
Striktně vzato, hodnota ::/0 značí dostupnost příchozího provozu pro server z celého světa, což není zcela správné, ale pro potřeby příkladu si tento přístup dovolíme použít:
Vracíme se na stránku prohlížeče, kde máme otevřené „Konfigurovat pokročilá nastavení“ a vybíráme v sekci Skupiny zabezpečení VPC —> Zvolit existující skupiny zabezpečení VPC —> PostgreSQL:
Dále, v sekci Možnosti databáze —> Název databáze —> nastavíme název — habrDB.
Ostatní parametry, s výjimkou snad vypnutí zálohování (doba uchování zálohy — 0 dní), monitorování a Performance Insights, můžeme ponechat jako výchozí. Klikneme na tlačítko Vytvořit databázi:
Zpracovatel proudů
Závěrečným krokem bude vývoj Spark úlohy, která každé dvě sekundy zpracovává nová data přijímaná od Kafka a ukládá výsledek do databáze.
Jak bylo zmíněno výše, kontrolní body (checkpoints) jsou hlavním mechanismem ve Spark Streaming, který musí být nastaven pro zajištění odolnosti. Budeme používat kontrolní body a v případě selhání procedury, modul Spark Streaming pro obnovení ztracených dat bude potřebovat pouze vrátit se k poslednímu kontrolnímu bodu a obnovit výpočty od něj.
Kontrolní bod lze zapnout nastavením adresáře v odolném, spolehlivém souborovém systému (například HDFS, S3 atd.), ve kterém budou uloženy informace kontrolního bodu. To se provádí například pomocí:
streamingContext.checkpoint(checkpointDirectory)V našem příkladu budeme používat následující přístup, a to, pokud checkpointDirectory existuje, kontext bude přenesen ze základních dat kontrolního bodu. Pokud adresář neexistuje (tj. provádí se poprvé), zavolá se funkce functionToCreateContext pro vytvoření nového kontextu a nastavení DStreams:
from pyspark.streaming import StreamingContext
context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)
Vytvoříme objekt DirectStream s cílem připojit se k tématu „transaction“ pomocí metody createDirectStream knihovny KafkaUtils:
z 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})
Parsneme příchozí data ve formátu JSON:
rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
currency=w['currency'],
amount=w['amount']))
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")
Pomocí Spark SQL provedeme jednoduchou agregaci a výstup zobrazíme v konzoli:
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
Získání textu dotazu a spuštění přes Spark SQL:
sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)
A poté uložíme získaná agregovaná data do tabulky v AWS RDS. Pro uložení výsledků agregace do databázové tabulky použijeme metodu write objektu DataFrame:
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()
Něco málo o nastavení připojení k AWS RDS. Uživatel a heslo jsme vytvořili v kroku „Nasazení AWS PostgreSQL“. Jako URL serveru databáze je třeba použít Endpoint, který se zobrazuje v části Connectivity & security:
Pro správné propojení Spark a Kafka je třeba spustit úlohu prostřednictvím spark-submit s použitím artefaktu spark-streaming-kafka-0-8_2.11. Dále použijeme také artefakt pro interakci s databází PostgreSQL a tyto budeme předávat prostřednictvím —packages.
Pro flexibilitu skriptu vyjmeme jako vstupní parametry také název serveru zpráv a topic, ze kterého chceme data získávat.
A teď nastal čas spustit a ověřit funkčnost systému:
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
Všechno funguje! Jak vidíte na obrázku níže — během provozu aplikace se nové výsledky agregace zobrazují každé 2 sekundy, protože jsme nastavili interval balíčkování na 2 sekundy při vytváření objektu StreamingContext:
Dále provedeme jednoduchý dotaz na databázi, abychom zkontrolovali přítomnost záznamů v tabulce transaction_flow:
Závěr
V tomto článku byl představen příklad zpracování informací v reálném čase pomocí Spark Streaming v kombinaci s Apache Kafka a PostgreSQL. S rostoucím objemem dat z různých zdrojů je obtížné přeceňovat praktickou hodnotu Spark Streaming pro vytváření streamovacích aplikací a aplikací fungujících v reálném čase.
Celý zdrojový kód můžete najít v mém repozitáři na .
Rád bych s vámi prodiskutoval tento článek, těším se na vaše komentáře a doufám v konstruktivní kritiku všech nezaujatých čtenářů.
Přeji hodně úspěchů!
Ps. Původně bylo plánováno použít místní databázi PostgreSQL, ale vzhledem k mé lásce k AWS jsem se rozhodl přesunout databázi do cloudu. V dalším článku na toto téma ukážu, jak realizovat výše popsaný systém v AWS pomocí AWS Kinesis a AWS EMR. Sledujte novinky!
Zdroj: habr.com
