Apache Kafka და სტრიმინგის მონაცემთა დამუშავება Spark Streaming-ით

გამარჯობა, ჰაბრ! დღეს ჩვენ ავაშენებთ სისტემას, რომელიც დაამუშავებს Apache Kafka-ს შეტყობინებების ნაკადებს Spark Streaming-ის გამოყენებით და ჩაწერს დამუშავების შედეგებს AWS RDS ღრუბლოვან მონაცემთა ბაზაში.

წარმოვიდგინოთ, რომ გარკვეული საკრედიტო დაწესებულება დასახავს ამოცანას, დავამუშაოთ შემომავალი ტრანზაქციები მის ყველა ფილიალში. ეს შეიძლება გაკეთდეს ხაზინის ღია ვალუტის პოზიციის, ტრანზაქციების ლიმიტების ან ფინანსური შედეგების და ა.შ.

როგორ განვახორციელოთ ეს საქმე ჯადოსნური და ჯადოსნური შელოცვების გამოყენების გარეშე - წაიკითხეთ ჭრილში! წადი!

(სურათის წყარო)

შესავალი

რა თქმა უნდა, დიდი რაოდენობით მონაცემების რეალურ დროში დამუშავება თანამედროვე სისტემებში გამოყენების უამრავ შესაძლებლობებს იძლევა. ამის ერთ-ერთი ყველაზე პოპულარული კომბინაციაა Apache Kafka-სა და Spark Streaming-ის ტანდემი, სადაც კაფკა ქმნის შემომავალი შეტყობინებების პაკეტების ნაკადს და Spark Streaming ამუშავებს ამ პაკეტებს მოცემულ დროის ინტერვალში.

აპლიკაციის შეცდომის ტოლერანტობის გასაზრდელად, ჩვენ გამოვიყენებთ საგუშაგოებს. ამ მექანიზმით, როდესაც Spark Streaming ძრავას სჭირდება დაკარგული მონაცემების აღდგენა, მას მხოლოდ ბოლო საგუშაგოზე დაბრუნება სჭირდება და იქიდან განაახლებს გამოთვლები.

განვითარებული სისტემის არქიტექტურა

გამოყენებული კომპონენტები:

• Apache Kafka არის განაწილებული გამოქვეყნება-გამოწერის შეტყობინებების სისტემა. ვარგისია როგორც ოფლაინ, ასევე ონლაინ შეტყობინებების მოხმარებისთვის. მონაცემთა დაკარგვის თავიდან ასაცილებლად, კაფკას შეტყობინებები ინახება დისკზე და მრავლდება კლასტერში. Kafka სისტემა აგებულია ZooKeeper სინქრონიზაციის სერვისის თავზე;
• Apache Spark Streaming - Spark კომპონენტი ნაკადის მონაცემების დასამუშავებლად. Spark Streaming მოდული აგებულია მიკრო-სერიული არქიტექტურის გამოყენებით, სადაც მონაცემთა ნაკადი ინტერპრეტირებულია, როგორც მცირე მონაცემთა პაკეტების უწყვეტი თანმიმდევრობა. Spark Streaming იღებს მონაცემებს სხვადასხვა წყაროდან და აერთიანებს მათ მცირე პაკეტებში. ახალი პაკეტები იქმნება რეგულარული ინტერვალებით. ყოველი დროის ინტერვალის დასაწყისში იქმნება ახალი პაკეტი და ამ ინტერვალის განმავლობაში მიღებული ნებისმიერი მონაცემი შედის პაკეტში. ინტერვალის ბოლოს, პაკეტის ზრდა ჩერდება. ინტერვალის ზომა განისაზღვრება პარამეტრით, რომელსაც ეწოდება პარტიული ინტერვალი;
• Apache Spark SQL - აერთიანებს ურთიერთობით დამუშავებას Spark-ის ფუნქციურ პროგრამირებასთან. სტრუქტურირებული მონაცემები ნიშნავს მონაცემებს, რომლებსაც აქვთ სქემა, ანუ ველების ერთი ნაკრები ყველა ჩანაწერისთვის. Spark SQL მხარს უჭერს შეყვანას სხვადასხვა სტრუქტურირებული მონაცემთა წყაროდან და, სქემის ინფორმაციის ხელმისაწვდომობის წყალობით, მას შეუძლია ეფექტურად მოიძიოს ჩანაწერების მხოლოდ საჭირო ველები და ასევე უზრუნველყოფს DataFrame API-ებს;
• AWS RDS არის შედარებით იაფი ღრუბელზე დაფუძნებული რელაციური მონაცემთა ბაზა, ვებ სერვისი, რომელიც ამარტივებს დაყენებას, ოპერაციას და მასშტაბირებას და ადმინისტრირებას ახდენს უშუალოდ Amazon-ის მიერ.

კაფკას სერვერის ინსტალაცია და გაშვება

კაფკას პირდაპირ გამოყენებამდე უნდა დარწმუნდეთ, რომ გაქვთ ჯავა, რადგან... JVM გამოიყენება სამუშაოდ:

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

მოდით შევქმნათ ახალი მომხმარებელი კაფკასთან სამუშაოდ:

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

შემდეგი, ჩამოტვირთეთ დისტრიბუცია Apache Kafka-ს ოფიციალური ვებსაიტიდან:

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

ჩამოტვირთული არქივის ამოლაგება:

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

შემდეგი ნაბიჯი არჩევითია. ფაქტია, რომ ნაგულისხმევი პარამეტრები არ გაძლევთ საშუალებას სრულად გამოიყენოთ Apache Kafka-ს ყველა შესაძლებლობა. მაგალითად, წაშალეთ თემა, კატეგორია, ჯგუფი, სადაც შეტყობინებები შეიძლება გამოქვეყნდეს. ამის შესაცვლელად, მოდით დაარედაქტიროთ კონფიგურაციის ფაილი:

vim ~/kafka/config/server.properties

ფაილის ბოლოს დაამატეთ შემდეგი:

delete.topic.enable = true

კაფკას სერვერის დაწყებამდე, თქვენ უნდა გაუშვათ ZooKeeper სერვერი; ჩვენ გამოვიყენებთ დამხმარე სკრიპტს, რომელიც მოყვება კაფკას განაწილებას:

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

მას შემდეგ, რაც ZooKeeper წარმატებით დაიწყო, გაუშვით Kafka სერვერი ცალკე ტერმინალში:

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

მოდით შევქმნათ ახალი თემა სახელწოდებით ტრანზაქცია:

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

მოდით დავრწმუნდეთ, რომ შეიქმნა თემა საჭირო რაოდენობის დანაყოფებითა და რეპლიკაციით:

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

გამოვტოვოთ მწარმოებლისა და მომხმარებლის ტესტირების მომენტები ახლად შექმნილი თემისთვის. დამატებითი დეტალები იმის შესახებ, თუ როგორ შეგიძლიათ შეამოწმოთ შეტყობინებების გაგზავნა და მიღება, იწერება ოფიციალურ დოკუმენტაციაში - გაგზავნეთ რამდენიმე შეტყობინება. კარგად, ჩვენ გადავდივართ პროდიუსერის დაწერაზე Python-ში KafkaProducer API-ის გამოყენებით.

პროდიუსერის წერა

პროდიუსერი გამოიმუშავებს შემთხვევით მონაცემებს - 100 შეტყობინებას ყოველ წამში. შემთხვევითი მონაცემებით ვგულისხმობთ ლექსიკონს, რომელიც შედგება სამი ველისაგან:

• ფილიალი - საკრედიტო დაწესებულების გაყიდვის პუნქტის დასახელება;
• ვალუტა — ტრანზაქციის ვალუტა;
• თანხა - გარიგების თანხა. თანხა იქნება დადებითი რიცხვი, თუ ეს არის ბანკის მიერ ვალუტის შეძენა და უარყოფითი რიცხვი, თუ ეს არის გაყიდვა.

პროდიუსერის კოდი ასე გამოიყურება:

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

შემდეგი, გაგზავნის მეთოდის გამოყენებით, ჩვენ ვუგზავნით შეტყობინებას სერვერზე, ჩვენთვის საჭირო თემაზე, 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('--> The message has been sent to a topic: 
            {}, partition: {}, offset: {}' 
            .format(record_metadata.topic,
                record_metadata.partition,
                record_metadata.offset ))   
                             
except Exception as e:
    print('--> It seems an Error occurred: {}'.format(e))

finally:
    producer.flush()

სკრიპტის გაშვებისას ტერმინალში ვიღებთ შემდეგ შეტყობინებებს:

ეს ნიშნავს, რომ ყველაფერი მუშაობს ისე, როგორც ჩვენ გვინდოდა - პროდიუსერი ქმნის და აგზავნის შეტყობინებებს ჩვენთვის საჭირო თემაზე.
შემდეგი ნაბიჯი არის Spark-ის დაყენება და ამ შეტყობინებების ნაკადის დამუშავება.

Apache Spark-ის ინსტალაცია

Apache Spark არის უნივერსალური და მაღალი ხარისხის კლასტერული გამოთვლითი პლატფორმა.

Spark უკეთესად ასრულებს MapReduce მოდელის პოპულარულ იმპლემენტაციას და მხარს უჭერს გამოთვლის ტიპების უფრო ფართო სპექტრს, მათ შორის ინტერაქტიული მოთხოვნებისა და ნაკადის დამუშავებას. სიჩქარე მნიშვნელოვან როლს ასრულებს დიდი რაოდენობით მონაცემების დამუშავებისას, რადგან ეს არის სიჩქარე, რომელიც საშუალებას გაძლევთ იმუშაოთ ინტერაქტიულად, წუთების ან საათების ლოდინის გარეშე. Spark-ის ერთ-ერთი ყველაზე ძლიერი მხარე, რაც მას ასე სწრაფს ხდის, არის მეხსიერებაში გამოთვლების შესრულების უნარი.

ეს ჩარჩო დაწერილია Scala-ში, ამიტომ ჯერ უნდა დააინსტალიროთ:

sudo apt-get install scala

ჩამოტვირთეთ Spark დისტრიბუცია ოფიციალური ვებსაიტიდან:

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

გახსენით არქივი:

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

დაამატეთ Spark-ის ბილიკი bash ფაილში:

vim ~/.bashrc

დაამატეთ შემდეგი სტრიქონები რედაქტორის საშუალებით:

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

გაუშვით ქვემოთ მოცემული ბრძანება bashrc-ში ცვლილებების შეტანის შემდეგ:

source ~/.bashrc

AWS PostgreSQL-ის დანერგვა

რჩება მხოლოდ მონაცემთა ბაზის განთავსება, რომელშიც ჩვენ ავტვირთავთ დამუშავებულ ინფორმაციას ნაკადებიდან. ამისათვის ჩვენ გამოვიყენებთ AWS RDS სერვისს.

გადადით AWS კონსოლზე -> AWS RDS -> მონაცემთა ბაზები -> მონაცემთა ბაზის შექმნა:

აირჩიეთ PostgreSQL და დააჭირეთ შემდეგი:

იმიტომ რომ ეს მაგალითი მხოლოდ საგანმანათლებლო მიზნებისთვისაა; ჩვენ გამოვიყენებთ უფასო სერვერს „მინიმუმ“ (უფასო დონე):

შემდეგი, ჩვენ ჩავსვამთ ტიკს Free Tier ბლოკში და ამის შემდეგ ავტომატურად შემოგვთავაზებენ t2.micro კლასის მაგალითს - თუმცა სუსტია, ის უფასოა და საკმაოდ შესაფერისია ჩვენი ამოცანის შესასრულებლად:

შემდეგ მოდის ძალიან მნიშვნელოვანი რამ: მონაცემთა ბაზის მაგალითის სახელი, მთავარი მომხმარებლის სახელი და მისი პაროლი. მოდით დავასახელოთ მაგალითი: myHabrTest, ძირითადი მომხმარებელი: ჰაბრ, პაროლი: habr12345 და დააჭირეთ ღილაკს შემდეგი:

შემდეგ გვერდზე არის პარამეტრები, რომლებიც პასუხისმგებელნი არიან ჩვენი მონაცემთა ბაზის სერვერის გარედან ხელმისაწვდომობაზე (Public accessibility) და პორტის ხელმისაწვდომობაზე:

მოდით შევქმნათ ახალი პარამეტრი VPC უსაფრთხოების ჯგუფისთვის, რომელიც საშუალებას მისცემს გარე წვდომას ჩვენს მონაცემთა ბაზის სერვერზე პორტით 5432 (PostgreSQL).
მოდით გადავიდეთ AWS კონსოლზე ცალკე ბრაუზერის ფანჯარაში VPC Dashboard -> უსაფრთხოების ჯგუფები -> შექმენით უსაფრთხოების ჯგუფის სექცია:

ჩვენ დავაყენეთ უსაფრთხოების ჯგუფის სახელი - PostgreSQL, აღწერა, მიუთითეთ რომელ VPC-თან უნდა იყოს დაკავშირებული ეს ჯგუფი და დააჭირეთ ღილაკს შექმნა:

შეავსეთ შემომავალი წესები 5432 პორტისთვის ახლად შექმნილი ჯგუფისთვის, როგორც ნაჩვენებია ქვემოთ მოცემულ სურათზე. თქვენ არ შეგიძლიათ ხელით მიუთითოთ პორტი, მაგრამ აირჩიეთ PostgreSQL ჩამოსაშლელი სიიდან Type.

მკაცრად რომ ვთქვათ, მნიშვნელობა ::/0 ნიშნავს სერვერზე შემომავალი ტრაფიკის ხელმისაწვდომობას მთელი მსოფლიოდან, რაც კანონიკურად მთლად სიმართლეს არ შეესაბამება, მაგრამ მაგალითის გასაანალიზებლად, მოდით საკუთარ თავს მივცეთ საშუალება გამოვიყენოთ ეს მიდგომა:

ჩვენ ვუბრუნდებით ბრაუზერის გვერდზე, სადაც გახსნილი გვაქვს “Configure advanced settings” და ვირჩევთ VPC უსაფრთხოების ჯგუფების განყოფილებაში -> აირჩიეთ არსებული VPC უსაფრთხოების ჯგუფები -> PostgreSQL:

შემდეგი, მონაცემთა ბაზის ოფციებში -> მონაცემთა ბაზის სახელი -> დააყენეთ სახელი - habrDB.

ჩვენ შეგვიძლია დარჩენილი პარამეტრები დავტოვოთ, გარდა სარეზერვო ასლის (სარეზერვო ასლის შენახვის ვადა - 0 დღე) გამორთვისა, მონიტორინგისა და Performance Insights, ნაგულისხმევად. დააჭირეთ ღილაკს მონაცემთა ბაზის შექმნა:

ძაფის დამმუშავებელი

დასკვნითი ეტაპი იქნება Spark job-ის შემუშავება, რომელიც ყოველ ორ წამში ერთხელ დაამუშავებს კაფკადან მოსულ ახალ მონაცემებს და შედეგს მონაცემთა ბაზაში შეიტანს.

როგორც ზემოთ აღინიშნა, საგუშაგოები არის SparkStreaming-ის ძირითადი მექანიზმი, რომელიც უნდა იყოს კონფიგურირებული შეცდომების ტოლერანტობის უზრუნველსაყოფად. ჩვენ გამოვიყენებთ საგუშაგოებს და თუ პროცედურა ვერ მოხერხდა, Spark Streaming მოდულს მხოლოდ ბოლო საგუშაგოზე დაბრუნება და მისგან გათვლების განახლება დასჭირდება დაკარგული მონაცემების აღსადგენად.

გამშვები პუნქტის ჩართვა შესაძლებელია შეცდომის ტოლერანტულ, საიმედო ფაილურ სისტემაზე (როგორიცაა HDFS, S3 და ა.შ.) დირექტორიაში დაყენებით, რომელშიც შეინახება საგუშაგოს ინფორმაცია. ეს კეთდება გამოყენებით, მაგალითად:

streamingContext.checkpoint(checkpointDirectory)

ჩვენს მაგალითში, ჩვენ გამოვიყენებთ შემდეგ მიდგომას, კერძოდ, თუ checkpointDirectory არსებობს, მაშინ კონტექსტი ხელახლა შეიქმნება საგუშაგოს მონაცემებიდან. თუ დირექტორია არ არსებობს (ანუ შესრულებულია პირველად), მაშინ ფუნქციაToCreateContext გამოიძახება ახალი კონტექსტის შესაქმნელად და DSstreams-ის კონფიგურაციისთვის:

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

ჩვენ ვქმნით DirectStream ობიექტს „ტრანზაქციის“ თემასთან დასაკავშირებლად KafkaUtils ბიბლიოთეკის 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})

შემომავალი მონაცემების გარჩევა JSON ფორმატში:

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

Spark SQL-ის გამოყენებით ვაკეთებთ მარტივ დაჯგუფებას და შედეგს ვაჩვენებთ კონსოლში:

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

მოთხოვნის ტექსტის მიღება და მისი გაშვება Spark SQL-ის მეშვეობით:

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

და შემდეგ ჩვენ ვინახავთ შედეგად აგრეგირებულ მონაცემებს ცხრილში AWS RDS-ში. აგრეგაციის შედეგების მონაცემთა ბაზის ცხრილში შესანახად, ჩვენ გამოვიყენებთ 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()

რამდენიმე სიტყვა AWS RDS-თან კავშირის დაყენების შესახებ. ჩვენ შევქმენით მისთვის მომხმარებელი და პაროლი "AWS PostgreSQL-ის დანერგვის" ეტაპზე. თქვენ უნდა გამოიყენოთ Endpoint, როგორც მონაცემთა ბაზის სერვერის url, რომელიც ნაჩვენებია დაკავშირების და უსაფრთხოების განყოფილებაში:

Spark-ისა და Kafka-ს სწორად დასაკავშირებლად, თქვენ უნდა შეასრულოთ სამუშაო smark-submit-ის საშუალებით, არტეფაქტის გამოყენებით. spark-streaming-kafka-0-8_2.11. გარდა ამისა, ჩვენ ასევე გამოვიყენებთ არტეფაქტს PostgreSQL მონაცემთა ბაზასთან ურთიერთობისთვის; ჩვენ მათ გადავცემთ --პაკეტების საშუალებით.

სკრიპტის მოქნილობისთვის, შეყვანის პარამეტრებად ასევე ჩავრთავთ შეტყობინებების სერვერის სახელს და თემას, საიდანაც გვინდა მონაცემების მიღება.

ასე რომ, დროა გაუშვათ და შეამოწმოთ სისტემის ფუნქციონირება:

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

ყველაფერი გამოვიდა! როგორც ქვემოთ სურათზე ხედავთ, სანამ აპლიკაცია მუშაობს, ახალი აგრეგაციის შედეგები გამოდის ყოველ 2 წამში, რადგან StreamingContext ობიექტის შექმნისას ჩვენ დავაყენეთ ჯგუფური ინტერვალი 2 წამზე:

შემდეგი, ჩვენ ვუკეთებთ მარტივ მოთხოვნას მონაცემთა ბაზაში, რათა შევამოწმოთ ჩანაწერების არსებობა ცხრილში გარიგების_ნაკადი:

დასკვნა

ეს სტატია განიხილავდა ინფორმაციის ნაკადის დამუშავების მაგალითს Spark Streaming-ის გამოყენებით Apache Kafka-სთან და PostgreSQL-თან ერთად. სხვადასხვა წყაროდან მონაცემების ზრდასთან ერთად, ძნელია გადაჭარბებული შეფასება Spark Streaming-ის პრაქტიკული მნიშვნელობის ნაკადის და რეალურ დროში აპლიკაციების შესაქმნელად.

თქვენ შეგიძლიათ იპოვოთ სრული წყარო კოდი ჩემს საცავში: GitHub.

სიამოვნებით განვიხილავ ამ სტატიას, ველოდები თქვენს კომენტარებს და ასევე იმედი მაქვს კონსტრუქციული კრიტიკის ყველა მზრუნველი მკითხველისგან.

გისურვებთ წარმატებას!

ფს. თავდაპირველად იგეგმებოდა ადგილობრივი PostgreSQL მონაცემთა ბაზის გამოყენება, მაგრამ AWS-ისადმი ჩემი სიყვარულის გათვალისწინებით, გადავწყვიტე მონაცემთა ბაზის ღრუბელში გადატანა. ამ თემაზე შემდეგ სტატიაში მე გაჩვენებთ, თუ როგორ უნდა განვახორციელოთ ზემოთ აღწერილი მთელი სისტემა AWS-ში AWS Kinesis-ისა და AWS EMR-ის გამოყენებით. მიჰყევით სიახლეებს!

წყარო: www.habr.com