🥇Apache Kafka ja andmete voogude töötlemine Spark Streaminguga

Tere, Habr! Täna ehitame süsteemi, mis töötleb Apache Kafka sõnumivoolusid Spark Streaming abil ja salvestab töötlemise tulemused AWS RDS pilveandmebaasi.

Kujutame ette, et mõni krediidiasutus esitab meile ülesande töötleda sissetulevaid tehinguid «reaalajas» kõigis oma filiaalides. Seda võib teha, et kiiresti arvutada avatud valuutapositsioon soovituslikule rahandusele, limiidid või finantsilised tulemused tehingute osas jne.

Kuidas seda juhtumit rakendada ilma maagia ja imeliste loitsudeta — loe edasi! Alustame!

(Pildi allikas)

Sissejuhatus

Ilmselgelt pakub suure andmemassi töötlemine reaalajas laialdasi võimalusi tänapäevastes süsteemides. Üks populaarsamaid kombinatsioone selleks on tandem Apache Kafka ja Spark Streaming, kus Kafka loob sissetulevate sõnumite voogusid ja Spark Streaming töötleb neid pakette määratud ajavahemiku jooksul.

Rakenduse kindluse suurendamiseks kasutame kontrollpunkte — checkpoints. Selle mehhanismi abil, kui Spark Streaming moodul peab taastama kadunud andmed, peab ta lihtsalt tagasipöörduma viimasest kontrollpunktist ja jätkama arvutusi sealt.

Arendatava süsteemi arhitektuur

Kasutatavad komponendid:

Apache Kafka — see on jagatud sõnumivahetussüsteem, mis toetab publikatsiooni ja tellimist. Sobib nii iseseisvaks kui ka veebipõhiseks sõnumite tarbimiseks. Andmete kadumise vältimiseks salvestatakse Kafka sõnumid kettale ja replitseeritakse klastris. Kafka süsteem on üles ehitatud ZooKeeperi sünkroonimisteenuse peale;
Apache Spark Streaming — Spark komponent voogandmete töötlemiseks. Spark Streaming moodul on üles ehitatud mikro-pakettide arhitektuurile, kus voogandmed tõlgendatakse kui pidevat väikeste andmepakettide järjestust. Spark Streaming võtab vastu andmeid erinevatest allikatest ja ühendab need väikeste pakkideks. Uued pakkide loomiseks kasutatakse regulaarselt ajaintervalli. Iga ajaintervalli alguses luuakse uus pakk, kuhu sisaldatakse kõik andmed, mis on saabunud selle intervalli jooksul. Intervalli lõpus lõpetatakse pakkide suurendamine. Intervalli suurus on määratud parameetriga, mida nimetatakse partii intervalliks.
Apache Spark SQL — ühendab relatsioonilise töötlemise Spark funktsionaalse programmeerimisega. Struktureerituna mõistetakse andmeid, millel on skeem, see tähendab ühtne väljade komplekt kõigikirjete jaoks. Spark SQL toetab mitmest struktureeritud andmete allikast sisendi, ja kuna skeemi teavet on olemas, saab see tõhusalt eraldada ainult vajalikud rekordite väljad ning pakub DataFrame API-sid.
AWS RDS — suhteliselt odav pilveline relatsiooniline andmebaas, veebiteenus, mis lihtsustab seadistamist, hooldust ja skaleerimist, mida haldab otse Amazon.

Kafka serveri paigaldamine ja käivitamine

Enne Kafka kasutamist tuleb veenduda, et Java on olemas, kuna kasutatakse JVM-i:

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

Loome uue kasutaja Kafka tööks:

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

Seejärel laadige alla distributsioon ametlikult Apache Kafka saidilt:

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

Pakkige alla laaditud arhiiv 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 täielikult kasutada Apache Kafka kõiki võimalusi. Näiteks ei saa kustutada teemasid, kategooriaid, gruppe, millele sõnumeid võib avaldada. Selle muutmiseks toimetame konfigureerimisfaili:

vim ~/kafka/config/server.properties

Lisage faili lõppu järgmine:

delete.topic.enable = true

Enne Kafka serveri käivitamist tuleb käivitada ZooKeeper server, kasutame abiskripti, mis on kaasas Kafka distributsiooniga:

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

Pärast seda, 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

Veenduge, et teema oleks loodud vajaliku arvu partitsioonide ja replikatsiooniga:

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

Jätame vahele testimise hetked produtsendi ja tarbija jaoks uue loodud teema puhul. Üksikasjalikumalt selle kohta, kuidas saatmist ja sõnumite vastuvõttu testida, on kirjas ametlikus dokumentatsioonis — Saada mõned sõnumid. Aga liigume edasi Pythonis produtsendi kirjutamise juurde, kasutades KafkaProducer API.

Produtsendi kirjutamine

Produtsent genereerib juhuslikke andmeid — 100 sõnumit iga sekundi jooksul. Juhuslike andmete all mõistame sõnastikku, mis koosneb kolmest valdkonnast:

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

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

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 sõnumi serverisse, soovitud teemas, 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('--> Sõnum on saadetud teemale: 
            {}, partitsioon: {}, positsioon: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> Tundub, 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 meie soovitud teemale.
Järgmine samm on Spark'i installeerimine ja selle sõnumivooge töötlemine.

Apache Spark'i paigaldamine

Apache Spark on universaalne ja kõrge jõudlusega klastritehnoloogia platvorm.

Jõudluse poolest ületab Spark populaarseid MapReduce mudeli rakendusi, pakkudes samal ajal laiemat toetust erinevat tüüpi arvutustele, sealhulgas interaktiivsetele päringutele ja voogedastusele. Kiirus mängib suurt rolli suurte andmemahtude töötlemisel, kuna just kiirus võimaldab töötada interaktiivselt, vältides minutite või tundide ootamist. Üks Spark'i suurimaid eeliseid, mis tagab sellise kõrge kiirus, on võime teostada arvutusi mälus.

Käesolev raamistik on kirjutatud Scala keeles, seetõttu tuleb see esmalt paigaldada:

sudo apt-get install scala

Laadime ametlikult Spark'i distributsiooni alla:

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

Pakkime arhiivi lahti:

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

Lisame tee Spark'i bash-faili:

vim ~/.bashrc

Lisa toimetajaga järgmised read:

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

Käivita allolev käsk pärast muudatuste tegemist bashrc-s:

source ~/.bashrc

AWS PostgreSQL'i rakendamine

Peame veel rakendama andmebaasi, kuhu laadime voogudest töödeldud teabe. Selleks kasutame AWS RDS teenust.

Logi sisse AWS konsooli —> AWS RDS —> Andmebaasid —> Loo andmebaas:

Valime PostgreSQL'i ja vajutame nuppu Järgmine:

Kuna seda näidet analüüsitakse ainult hariduslikel eesmärkidel, kasutame tasuta serverit "minimaalses suuruses" (Free Tier):

Seejärel märkime Free Tier ploki, ja pärast seda pakutakse meile automaatselt t2.micro instantsi — kuigi see on nõrk, on see tasuta ja sobib meie ülesande jaoks hästi:

Järgmised on väga olulised asjad: andmebaasi instantsi nimi, administraator-kasutaja nimi ja tema parool. Nimetame instantsi: myHabrTest, administraator-kasutaja: habr, parool: habr12345 ja vajutame nuppu Järgmine:

Järgmisel lehel on seaded, mis vastutavad meie andmebaasi serveri väliskäideldavuse (Avalik ligipääs) ja portide kättesaadavuse eest:

Loome VPC turvagruppide jaoks uue konfiguratsiooni, mis võimaldab väljastpoolt meie andmebaasi serverisse pääseda kaudu pordi 5432 (PostgreSQL).
Liigume teises brauseriaknas AWS konsooli VPC armatuurlauale —> Turvagruppide haldamine —> Loo turvagrupp:

Määrame nime Security group – PostgreSQL, kirjeldame seda, täpsustame, millise VPC-ga see grupp peab olema seotud, ja vajutame nuppu Create:

Täidame värskelt loodud grupile Inbound rules sadamale 5432, nagu on näidatud alloleval pildil. Sadamat ei ole vaja käkitehnikat määrata, vaid valida PostgreSQL rippmenüüst Type.

Tegelikult tähendab väärtus ::/0 juurdepääsu sissetulevale liiklusele serverile kogu maailmast, mis ei ole täiesti õige, kuid näite mõistmiseks lubame endale sellist lähenemist:

Naaseme brauseri lehele, kus meil on avatud "Configure advanced settings" ja valime jaotisest VPC security groups –> Choose existing VPC security groups –> PostgreSQL:

Seejärel, jaotises Database options –> Database name –> määrame nime – habrDB.

Ülejäänud parameetrid, välja arvatud võib-olla varukoopiate keelamine (backup retention period – 0 days), jälgimine ja Performance Insights, saame jätta vaikeväärtustele. Vajutame nuppu Create database:

Voogude töötleja

Lõppfaasiks on Spark-i töö arendamine, mis igal kahel sekundil töötleb uusi andmeid, mis on arrivedud Kafka kaudu ja salvestab tulemuse andmebaasi.

Nagu eespool mainitud, on kontrollpunktid (checkpoints) põhimehhanism SparkStreamingus, mis peab olema seadistatud talitluse katkemise vältimiseks. Kasutame kontrollpunkte ja juhul, kui protseduur kukub kokku, peab Spark Streaming moodul kaotatud andmete taastamiseks lihtsalt naasma viimasele kontrollpunktile ja jätkama kalkulatsioone sellelt.

Kontrollpunkti saab lubada, määrates katalooge usaldusväärses, väljatõrjutavas failisüsteemis (nt HDFS, S3 jne), kus kontrollpunkti teave salvestatakse. Seda tehakse näiteks:

streamingContext.checkpoint(checkpointDirectory)

Meie näites kasutame järgmist lähenemist, nimelt kui checkpointDirectory eksisteerib, siis konteksti taastatakse kontrollpunkti andmetest. Kui kataloog ei eksisteeri (st see käivitatakse esmakordselt), siis kutsutakse välja funktsioon functionToCreateContext uue konteksti loomiseks ja DStreamide seadistamiseks:

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

Loome objekti DirectStream, et ühendada teema «transaction» abil meetodi createDirectStream kaudu, kasutades KafkaUtils-i:

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

Parsime sisenevaid 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-i, teeme lihtsa rühmitamise ja kuvame tulemuse konsoolile:

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

Saame päringu teksti ja käivitame selle Spark SQL-is:

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

Seejärel salvestame saadud agrigeeritud andmed tabelisse AWS RDS. Tulemuste salvestamiseks andmebaasitabelisse kasutame DataFrame'i 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()

Mõned sõnad AWS RDS-iga ühenduse seadistamisest. Kasutaja ja parooli lõime me "AWS PostgreSQL-i juurutamise" etapis. Andmebaasi serveri url-ina tuleks kasutada Endpoint'i, mis kuvatakse Connectivity & security jaotises:

Selleks, et Spark ja Kafka õigesti koos töötaksid, tuleks töö käivitada spark-submit kaudu artefakti abil spark-streaming-kafka-0-8_2.11. Samuti rakendame PostgreSQL-i andmebaasi suhtlemiseks vajalikku artefakti, mille edastame läbi —packages.

Skripti paindlikkuse tagamiseks toome ka sõnumiserveri nime ja teema, kust soovime andmeid saada, sisse sisendparameetriteks.

Nii et on aeg süsteem käivitada ja selle töökindlust 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 õnnestus! Nagu allolevalt pildilt näha — rakenduse töötamise ajal kuvatakse uusi agrigeeritud tulemusi iga 2 sekundi järel, kuna seadsime paketide intervalli 2 sekundiks, kui loome StreamingContext objekti:

Seejärel teeme lihtsa päringu andmebaasi, et kontrollida tabelis salvestuste olemasolu transaction_flow:

Kokkuvõte

Selles artiklis käsitletakse näidet andmevoo töötlemisest, kasutades Spark Streaming koos Apache Kafka ja PostgreSQL-iga. Andmemahtude suurenedes erinevatest allikatest on raske üle hinnata Spark Streaming'i praktilist väärtust voograkenduste ja reaalajas töötavate rakenduste loomisel.

Kogu algkoodi leiate minu repolt aadressilt GitHub.

Olen hea meelega valmis arutama seda artiklit, ootan teie kommentaare ja loodan konstruktiivset kriitikat kõigilt, kes on huvitatud.

Soovin edu!

Ps. Alguses oli plaanis kasutada kohalikke PostgreSQL andmebaase, kuid arvestades minu armastust AWS-i vastu, otsustasin viia andmebaasi pilve. Järgmises artiklis selle teema kohta näitan, kuidas rakendada täielikult ülaltoodud süsteemi AWS-is, kasutades AWS Kinesis ja AWS EMR. Jälgige uudiseid!

Allikas: habr.com