Apache Kafka ja andmevoogude töötlemine Spark Streaminguga

Tere, Habr! Täna ehitame süsteemi, mis kasutab Spark Streamingut Apache Kafka sõnumivoogude töötlemiseks ja kirjutab töötlemise tulemuse AWS RDS pilveandmebaasi.

Kujutage ette, et vahendite ettevõte seab meile ülesande sisenevate tehingute töötlemiseks „reaalajas“ kõigis oma filiaalides. Seda võib teha, et kiiresti arvutada avatud valuuta positsiooni, limiteid või finantstulemust tehingute osas jne.

Kuidas seda juhtumit ellu viia ilma maagia ja nõiduse rakendamiseta — loe edasi! Alustame!

(Pildi allikas)

Sissejuhatus

Ilmselt pakub suure hulga andmete töötlemine reaalajas kaasaegsetes süsteemides laialdasi võimalusi. Üheks populaarseks kombinatsiooniks on Apache Kafka ja Spark Streaming tandem, kus Kafka genereerib sisenevate sõnumite voogu ja Spark Streaming töötleb neid pakette määratud ajavahemiku jooksul.

Rakenduse seisukindluse suurendamiseks kasutame kontrollpunkte — checkpoints. Selle mehhanismi abil, kui Spark Streaming moodul vajab kadunud andmete taastamist, peab see lihtsalt naasma viimasest kontrollpunktist ja jätkama arvutusi sealt.

Arendatava süsteemi arhitektuur

Kasutatavad komponendid:

• Apache Kafka — see on hajustatud sõnumite jagamise süsteem, millel on avaldamine ja tellimine. Sobib nii iseseisvaks kui ka veebipõhiseks sõnumite tarbimiseks. Andmete kaotuse vältimiseks salvestatakse Kafka sõnumid kettale ja replikeeritakse klastris. Kafka süsteem on üles ehitatud ZooKeeper sünkroniseerimisteenuse peale;
• Apache Spark Streaming — Spark komponent andmete voogude töötlemiseks. Spark Streaming moodul on üles ehitatud mikropartiidest koosneva arhitektuuri alusel, kus andmevoog tõlgendatakse pideva väikeste andmepakettide jadana. Spark Streaming võtab andmeid erinevatest allikatest ja ühendab need väikesteks pakettideks. Uued paketid luuakse regulaarsete ajavahemike järel. Iga ajavahemiku alguses luuakse uus pakett, ja kõik andmed, mis selle ajavahemiku jooksul saabuvad, lisatakse paketti. Ajavahemiku lõpus paketise suurendamine lõppeb. Ajavahemiku suurus määratakse parameetriga nimega paketi intervall;
• Apache Spark SQL — ühendab relatsioonilise töötlemise Spark'i funktsionaalse programmeerimisega. Struktureeritud andmed viitavad andmetele, millel on skeem, st kõigi kirje jaoks ühtne väljade kogum. Spark SQL toetab sisendit mitmest struktureeritud andmete allikast ja skeeminformatsiooni olemasolu tõttu suudab tõhusalt ekstraktsioonida ainult vajalikud väljad, samuti pakub DataFrame API.
• AWS RDS — see on suhteliselt odav pilvepõhine relatsiooniline andmebaas, veebiteenus, mis lihtsustab seadistamist, haldamist ja skaleerimist ning mida haldab otse Amazon.

Kafka serveri paigaldamine ja käivitamine

Enne Kafka kasutamist peate veenduma, et Java on olemas, kuna JVM-i kasutatakse tööks:

sudo apt-get update 
sudo apt-get install default-jre
java -version

Loome uue kasutaja Kafka jaoks:

sudo useradd kafka -m
sudo passwd kafka
sudo adduser kafka sudo

Seejärel laadime alla distributsiooni Apache Kafka ametlikult veebisaidilt:

wget -P /YOUR_PATH "http://apache-mirror.rbc.ru/pub/apache/kafka/2.2.0/kafka_2.12-2.2.0.tgz"

Pakkime alla laetud arhiivi lahti:

tar -xvzf /YOUR_PATH/kafka_2.12-2.2.0.tgz
ln -s /YOUR_PATH/kafka_2.12-2.2.0 kafka

Järgmine samm on valikuline. Asjaolu on see, et vaikeseaded ei võimalda Apache Kafka kõiki funktsioone täielikult kasutada. Näiteks teema, kategooria või grupi kustutamine, kuhu sõnumeid avaldada. Selle muutmiseks redigeerime konfiguratsioonifaili:

vim ~/kafka/config/server.properties

Lisage faili lõppu järgmine:

delete.topic.enable = true

Enne Kafka serveri käivitamist tuleb käivitada ZooKeeperi server. Kasutame abiskripti, mis tuleb koos Kafka distributsiooniga:

Cd ~/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

Kui ZooKeeper on edukalt käivitunud, käivitame Kafka serveri eraldi terminalis:

bin/kafka-server-start.sh config/server.properties

Loome uue teema nimega Transaction:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

Veendume, et teema on loodud vajaliku partitsioonide arvu ja replikatsiooniga:

bin/kafka-topics.sh --describe --zookeeper localhost:2181

Jätame vahele proovide testimise tootja ja tarbija jaoks uue teema. Täpsemalt, kuidas saata ja vastu võtta sõnumeid, on kirjas ametlikus dokumentatsioonis — Saada mõned sõnumid. Siiski, liigume edasi Pythonis tootja kirjutamise juurde, kasutades KafkaProducer API-t.

Tootja kirjutamine

Tootja genereerib juhuslikke andmeid — 100 sõnumit iga sekundi tagant. Juhuslike andmete all mõistame kolme välja koosnevat sõnastikku:

• Filiaal — krediidiasutuse müügikoha nimi;
• Currency — tehingu valuuta;
• Amount — tehingu summa. Summa on positiivne number, kui see on valuuta ostmine pangast, ja negatiivne, kui müük.

Produtsendi kood näeb välja järgmiselt:

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

Seejärel, kasutades meetodit send, saadame teate serverile, sobivasse teema, JSON formaadis:

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('--> Teade on saadetud teemasse: 
            {}, jagu: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> Näib, et tekkis viga: {}'.format(e))

finally:
    producer.flush()

Skripti käivitamisel saame terminalis järgmised sõnumid:

See tähendab, et kõik töötab nii nagu soovisime — produtsent genereerib ja saadab sõnumeid sobivasse teema.
Järgmine samm on Spark'i installimine ja selle sõnumivoogude töötlemine.

Apache Spark'i installimine

Apache Spark on universaalne ja suure jõudlusega klastritehnoloogia arvutusalus.

Spark ületab oma jõudluses tuntud MapReduce'i rakendused, pakkudes samal ajal toe laiema hulga arvutusliikide, sealhulgas interaktiivsete päringute ja voogedastuse jaoks. Kiirus mängib suurt rolli suurte andmemahtude töötlemisel, kuna just kiirus võimaldab töötada interaktiivses režiimis, raiskamata minuteid või tunde ootamisele. Üks Spark'i peamisi eeliseid, mis tagab sellise kõrge kiirus, on võime teostada arvutusi mälus.

See raamistik on kirjutatud Scala keeles, seega tuleb see esmalt installida:

sudo apt-get install scala

Laadime ametlikult alla Spark'i distributsiooni:

wget "http://mirror.linux-ia64.org/apache/spark/spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz"

Pakkime arhivi lahti:

sudo tar xvf spark-2.4.2/spark-2.4.2-bin-hadoop2.7.tgz -C /usr/local/spark

Lisame Spark'i tee bash-faili:

vim ~/.bashrc

Lisame redigeerijaga järgmised read:

SPARK_HOME=/usr/local/spark
export PATH=$SPARK_HOME/bin:$PATH

Käivitage alljärgnev käsk pärast muutmisi bashrc-s:

source ~/.bashrc

AWS PostgreSQLi üles seadmine

Järgmiseks peame üles seadma andmebaasi, kuhu laadime voogudest töödeldud teabe. Selleks kasutame AWS RDS teenust.

Logige sisse AWS konsooli —> AWS RDS —> Andmebaasid —> Looge andmebaas:

Valige PostgreSQL ja klõpsake nuppu Järgmine:

Arvestades, et see näide on mõeldud ainult hariduslikel eesmärkidel, kasutame tasuta serverit 'minimaalsel tasemel' (Free Tier):

Seejärel märkige Free Tier kast ja pärast seda pakutakse automaatselt instantsi klassist t2.micro — kuigi see on nõrk, on see tasuta ja sobib meie vajadustele:

Seejärel on väga olulised viisid: andmebaasi instantsi nimi, master-kasutaja nimi ja selle parool. Nimetame instantsi: myHabrTest, master-kasutaja: habr, parool: habr12345 ja klõpsake nuppu Järgmine:

Järgmises lehes on parameetrid, mis vastutavad meie andmebaasi serveri välise kättesaadavuse (Public accessibility) ja portsede kättesaadavuse eest:

Loo uus seadistus VPC turvagruppi, mis võimaldab väljastpoolt meie andmebaasi serveriga ühendust luua sadamast 5432 (PostgreSQL).
Avame eraldi brauseriaknas AWS konsoolis VPC juhtpaneelis ja valime turvagruppide alt Create security group:

Kasutame turvagrupi jaoks nime — PostgreSQL, lisame kirjelduse, valime, mille VPC-ga see grupp peab olema seotud, ja vajutame nuppu Create:

Täidame uue grupi Inbound rules reeglid portidele 5432, nagu on näidatud alloleval pildil. Porti ei pea käsitsi määrama, vaid valige PostgreSQL rippmenüüst Type.

Täpsemalt öeldes tähendab väärtus ::/0, et sisenev liiklus on serverile saadaval üle kogu maailma, mis pole klassikaliselt täiesti õige, kuid selle näite analüüsimiseks lubame endale sellise lähenemise:

Naaseme brauseri lehele, kus on avatud „Configure advanced settings“ ja valime VPC turvagruppide alt — Choose existing VPC security groups — PostgreSQL:

Seejärel Database options — Database name — lisame nime — habrDB.

Ülejäänud parameetreid, välja arvatud varundamise väljalülitamine (backup retention period — 0 days), jälgimise ja Performance Insights osas, võime jätta vaikimisi. Vajutame nuppu Create database:

Voogude töötleja

Viimane etapp on Spark-töötluse arendamine, mis töötleb iga kahe sekundi järel uusi andmeid, mis saabuvad Kafka'st, ja talletab tulemuse andmebaasi.

Nagu eespool mainitud, on kontrollpunktid (checkpoints) põhiline mehhanism SparkStreaming'is, mis tuleb seadistada, et tagada süsteemi vastupidavus. Kasutame kontrollpunkte ja juhul, kui protseduur tõrkub, võib Spark Streaming mooduli abil kadunud andmed taastada, lihtsalt naastes viimasele kontrollpunktile ja jätkates arvutusi sealt.

Kontrollpunkti saab aktiveerida, määrates katalooge vastupidavasse, usaldusväärsesse failisüsteemi (nt HDFS, S3 jne), kuhu salvestatakse kontrollpunkti teave. Seda saab teha näiteks:

streamingContext.checkpoint(checkpointDirectory)

Meie näites rakendame järgmist lähenemist, nimelt, kui checkpointDirectory eksisteerib, rekonstrueeritakse kontekst kontrollpunkti andmete põhjal. Kui kataloog ei eksisteeri (st see toimub esmakordselt), siis kutsutakse välja funktsioon functionToCreateContext uue konteksti loomiseks ja DStreamide seadistamiseks:

import pyspark.streaming.StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

Loome objekti DirectStream, et ühendada teema 'transaction' kasutades KafkaUtils raamatukogu meetodit createDirectStream:

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})

Parseerime sissetulevaid andmeid JSON formaadis:

rowRdd = rdd.map(lambda w: Row(branch=w['branch'],
                                       currency=w['currency'],
                                       amount=w['amount']))
                                       
testDataFrame = spark.createDataFrame(rowRdd)
testDataFrame.createOrReplaceTempView("treasury_stream")

Kasutades Spark SQL, teeme lihtsa grupeerimise ja prindime tulemuse konsooli:

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

Küsimuse teksti saamine ja selle Käitamine Spark SQL-is:

sql_query = get_sql_query()
testResultDataFrame = spark.sql(sql_query)
testResultDataFrame.show(n=5)

Ja seejärel salvestame saadud koondatud andmed tabelisse AWS RDS. Andmete salvestamiseks kasutame DataFrame objekti meetodit write:

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()

AWS RDS-iga ühenduse seadistamisest. Kasutajat ja parooli loodi sammul "AWS PostgreSQL-i juurutamine". Andmebaasi serveri URL-ina tuleks kasutada Endpoint'i, mis kuvatakse Connectivity & security jaotises:

Selleks, et Spark ja Kafka korrektselt suhelda, tuleks töö üles laadida smark-submit kaudu, kasutades artefakti spark-streaming-kafka-0-8_2.11. Täiendavalt rakendame ka artefakti PostgreSQL andmebaasiga suhtlemiseks, edastame need läbi —packages.

Skripti paindlikkuse huvides, toome sisse ka sõnumite serveri ja teema nimed, millest soovime andmeid saada.

Nüüd on aeg süsteemi käivitada ja selle töökorrasolekut kontrollida:

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

Kõik läks korda! Nagu allolevalt pildilt näha — rakenduse töö käigus kuvatakse uusi agregatsioonitulemusi iga 2 sekundi tagant, kuna seadsime paketi intervalli 2 sekundile, kui lõime StreamingContext objekti:

Järgmisena teeme lihtsa päringu andmebaasi, et kontrollida, kas tabelis on kirjeid transaction_flow:

Kokkuvõte

Antud artiklis käsitleti näidet voogedastuse töötlemisest Spark Streaming kasutamisega koos Apache Kafka ja PostgreSQL-ga. Andmemahtude kasvu tõttu erinevatest allikatest on keeruline üle hinnata Spark Streaming praktilist väärtust voogedastusrakenduste ja reaalajas toimivate rakenduste loomisel.

Täieliku lähtekoodi leiate minu repolt GitHub.

Olen valmis arutama seda artiklit, ootan teie kommentaare ning loodan kõigi murelike lugejate konstruktiivset kriitikat.

Soovin edu!

Ps. Alguses plaaniti kasutada kohalikku PostgreSQL andmebaasi, kuid arvestades minu armastust AWS-i vastu, otsustasin viia andmebaasi pilve. Järgmises artiklis sellel teemal näitan, kuidas ellu viia täielikult ülal kirjeldatud süsteem AWS-is, kasutades AWS Kinesis ja AWS EMR. Hoidke end kursis!

Allikas: habr.com