🥇Apache Kafka आणि Spark Streaming वापरून स्ट्रीमिंग डेटा प्रोसेसिंग

हॅलो, हॅब्र! आज आम्ही एक प्रणाली तयार करू जी स्पार्क स्ट्रीमिंग वापरून Apache Kafka संदेश प्रवाहांवर प्रक्रिया करेल आणि प्रक्रिया परिणाम AWS RDS क्लाउड डेटाबेसवर लिहू.

चला कल्पना करूया की एखादी विशिष्ट क्रेडिट संस्था आपल्या सर्व शाखांमध्ये “ऑन द फ्लाय” इनकमिंग व्यवहारांवर प्रक्रिया करण्याचे कार्य सेट करते. खजिन्यासाठी खुल्या चलनाची स्थिती, व्यवहारांसाठी मर्यादा किंवा आर्थिक परिणाम इत्यादींची त्वरित गणना करण्याच्या हेतूने हे केले जाऊ शकते.

जादू आणि जादूच्या मंत्रांचा वापर न करता या केसची अंमलबजावणी कशी करावी - कट अंतर्गत वाचा! जा!

(प्रतिमा स्त्रोत)

परिचय

अर्थात, रिअल टाइममध्ये मोठ्या प्रमाणात डेटावर प्रक्रिया केल्याने आधुनिक प्रणालींमध्ये वापरासाठी भरपूर संधी उपलब्ध होतात. यासाठी सर्वात लोकप्रिय संयोजनांपैकी एक म्हणजे Apache Kafka आणि Spark Streaming चे टॅन्डम, जिथे काफ्का येणार्‍या संदेश पॅकेट्सचा एक प्रवाह तयार करतो आणि स्पार्क स्ट्रीमिंग दिलेल्या वेळेच्या अंतराने या पॅकेट्सवर प्रक्रिया करते.

अनुप्रयोगाची दोष सहिष्णुता वाढवण्यासाठी, आम्ही चेकपॉइंट्स वापरू. या यंत्रणेसह, जेव्हा स्पार्क स्ट्रीमिंग इंजिनला गमावलेला डेटा पुनर्प्राप्त करण्याची आवश्यकता असते, तेव्हा त्याला फक्त शेवटच्या चेकपॉईंटवर परत जाणे आणि तेथून गणना पुन्हा सुरू करणे आवश्यक आहे.

विकसित प्रणालीचे आर्किटेक्चर

वापरलेले घटक:

अपाचे काफ्का एक वितरित प्रकाशन-सदस्यता संदेश प्रणाली आहे. ऑफलाइन आणि ऑनलाइन संदेश वापरासाठी योग्य. डेटा गमावण्यापासून रोखण्यासाठी, काफ्का संदेश डिस्कवर संग्रहित केले जातात आणि क्लस्टरमध्ये प्रतिरूपित केले जातात. झूकीपर सिंक्रोनाइझेशन सेवेच्या शीर्षस्थानी काफ्का प्रणाली तयार केली गेली आहे;
अपाचे स्पार्क प्रवाह - स्ट्रीमिंग डेटावर प्रक्रिया करण्यासाठी स्पार्क घटक. स्पार्क स्ट्रीमिंग मॉड्यूल मायक्रो-बॅच आर्किटेक्चर वापरून तयार केले आहे, जेथे डेटा प्रवाहाचा अर्थ लहान डेटा पॅकेट्सचा सतत क्रम म्हणून केला जातो. स्पार्क स्ट्रीमिंग विविध स्त्रोतांकडून डेटा घेते आणि लहान पॅकेजेसमध्ये एकत्र करते. नवीन पॅकेजेस नियमित अंतराने तयार केले जातात. प्रत्येक वेळेच्या मध्यांतराच्या सुरुवातीला, एक नवीन पॅकेट तयार केले जाते आणि त्या मध्यांतरादरम्यान प्राप्त झालेला कोणताही डेटा पॅकेटमध्ये समाविष्ट केला जातो. मध्यांतराच्या शेवटी, पॅकेटची वाढ थांबते. मध्यांतराचा आकार बॅच इंटरव्हल नावाच्या पॅरामीटरद्वारे निर्धारित केला जातो;
अपाचे स्पार्क एसक्यूएल - स्पार्क फंक्शनल प्रोग्रामिंगसह रिलेशनल प्रोसेसिंग एकत्र करते. स्ट्रक्चर्ड डेटा म्हणजे स्कीमा असलेला डेटा, म्हणजेच सर्व रेकॉर्डसाठी फील्डचा एकच संच. स्पार्क एसक्यूएल विविध संरचित डेटा स्त्रोतांकडून इनपुटला समर्थन देते आणि, स्कीमा माहितीच्या उपलब्धतेबद्दल धन्यवाद, ते केवळ रेकॉर्डच्या आवश्यक फील्ड कार्यक्षमतेने पुनर्प्राप्त करू शकते आणि डेटाफ्रेम 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

पुढे, अधिकृत अपाचे काफ्का वेबसाइटवरून वितरण डाउनलोड करा:

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

पुढील चरण वैकल्पिक आहे. वस्तुस्थिती अशी आहे की डीफॉल्ट सेटिंग्ज आपल्याला अपाचे काफ्काच्या सर्व क्षमता पूर्णपणे वापरण्याची परवानगी देत नाहीत. उदाहरणार्थ, एखादा विषय, श्रेणी, गट हटवा ज्यावर संदेश प्रकाशित केले जाऊ शकतात. हे बदलण्यासाठी, कॉन्फिगरेशन फाइल संपादित करूया:

vim ~/kafka/config/server.properties

फाईलच्या शेवटी खालील जोडा:

delete.topic.enable = true

काफ्का सर्व्हर सुरू करण्यापूर्वी, तुम्हाला ZooKeeper सर्व्हर सुरू करणे आवश्यक आहे; आम्ही काफ्का वितरणासह येणारी सहाय्यक स्क्रिप्ट वापरू:

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

ZooKeeper यशस्वीरित्या सुरू झाल्यानंतर, वेगळ्या टर्मिनलमध्ये काफ्का सर्व्हर लाँच करा:

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

नवीन तयार केलेल्या विषयासाठी उत्पादक आणि ग्राहकांच्या चाचणीचे क्षण गमावूया. आपण संदेश पाठवणे आणि प्राप्त करणे कसे तपासू शकता याबद्दल अधिक तपशील अधिकृत कागदपत्रांमध्ये लिहिलेले आहेत - काही संदेश पाठवा. बरं, आम्ही KafkaProducer API वापरून Python मध्ये निर्माता लिहिण्याकडे पुढे जाऊ.

निर्माता लेखन

निर्माता यादृच्छिक डेटा व्युत्पन्न करेल - प्रत्येक सेकंदाला 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()

स्क्रिप्ट चालवताना, आम्हाला टर्मिनलमध्ये खालील संदेश प्राप्त होतात:

याचा अर्थ असा की सर्वकाही आम्हाला हवे तसे कार्य करते - निर्माता आम्हाला आवश्यक असलेल्या विषयावर संदेश तयार करतो आणि पाठवतो.
पुढील पायरी म्हणजे स्पार्क स्थापित करणे आणि या संदेश प्रवाहावर प्रक्रिया करणे.

अपाचे स्पार्क स्थापित करत आहे

अपाचे स्पार्क एक सार्वत्रिक आणि उच्च-कार्यक्षमता क्लस्टर संगणकीय प्लॅटफॉर्म आहे.

इंटरएक्टिव्ह क्वेरी आणि स्ट्रीम प्रोसेसिंगसह, संगणकीय प्रकारांच्या विस्तृत श्रेणीचे समर्थन करताना, मॅपरेड्यूस मॉडेलच्या लोकप्रिय अंमलबजावणीपेक्षा स्पार्क चांगली कामगिरी करते. मोठ्या प्रमाणात डेटावर प्रक्रिया करताना गती महत्त्वाची भूमिका बजावते, कारण ही गती आहे जी तुम्हाला काही मिनिटे किंवा तास प्रतीक्षा न करता परस्पररित्या कार्य करण्यास अनुमती देते. स्पार्कचे सर्वात मोठे सामर्थ्य जे ते इतके जलद बनवते ती म्हणजे मेमरीमधील गणना करण्याची क्षमता.

हे फ्रेमवर्क स्कालामध्ये लिहिलेले आहे, म्हणून तुम्हाला ते प्रथम स्थापित करणे आवश्यक आहे:

sudo apt-get install scala

अधिकृत वेबसाइटवरून स्पार्क वितरण डाउनलोड करा:

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

बॅश फाइलमध्ये स्पार्कचा मार्ग जोडा:

vim ~/.bashrc

संपादकाद्वारे खालील ओळी जोडा:

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

bashrc मध्ये बदल केल्यानंतर खालील आदेश चालवा:

source ~/.bashrc

AWS PostgreSQL तैनात करत आहे

फक्त डेटाबेस तैनात करणे बाकी आहे ज्यामध्ये आम्ही प्रवाहांमधून प्रक्रिया केलेली माहिती अपलोड करू. यासाठी आम्ही AWS RDS सेवा वापरू.

AWS कन्सोलवर जा -> AWS RDS -> डेटाबेस -> डेटाबेस तयार करा:

PostgreSQL निवडा आणि पुढील क्लिक करा:

कारण हे उदाहरण केवळ शैक्षणिक उद्देशांसाठी आहे; आम्ही "किमान" (विनामूल्य श्रेणी) विनामूल्य सर्व्हर वापरू.

पुढे, आम्ही फ्री टियर ब्लॉकमध्ये एक टिक ठेवतो आणि त्यानंतर आम्हाला t2.micro वर्गाचे उदाहरण आपोआप ऑफर केले जाईल - जरी कमकुवत असले तरी ते विनामूल्य आहे आणि आमच्या कार्यासाठी योग्य आहे:

पुढे खूप महत्त्वाच्या गोष्टी येतात: डेटाबेस उदाहरणाचे नाव, मुख्य वापरकर्त्याचे नाव आणि त्याचा पासवर्ड. चला उदाहरणाचे नाव देऊ: myHabrTest, मास्टर वापरकर्ता: habr, पासवर्ड: habr12345 आणि पुढील बटणावर क्लिक करा:

पुढील पृष्ठावर आमच्या डेटाबेस सर्व्हरच्या बाहेरून प्रवेशयोग्यता (सार्वजनिक प्रवेशयोग्यता) आणि पोर्ट उपलब्धतेसाठी जबाबदार पॅरामीटर्स आहेत:

चला VPC सुरक्षा गटासाठी एक नवीन सेटिंग तयार करू, जे पोर्ट 5432 (PostgreSQL) द्वारे आमच्या डेटाबेस सर्व्हरवर बाह्य प्रवेशास अनुमती देईल.
वेगळ्या ब्राउझर विंडोमध्ये AWS कन्सोलवर VPC डॅशबोर्ड -> सुरक्षा गट -> सुरक्षा गट विभाग तयार करा:

आम्ही सुरक्षा गटासाठी नाव सेट केले आहे - पोस्टग्रेएसक्यूएल, वर्णन, हा गट कोणत्या व्हीपीसीशी संबंधित असावा आणि तयार करा बटण क्लिक करा:

खालील चित्रात दाखवल्याप्रमाणे, नव्याने तयार केलेल्या गटासाठी पोर्ट 5432 साठी इनबाउंड नियम भरा. तुम्ही पोर्ट व्यक्तिचलितपणे निर्दिष्ट करू शकत नाही, परंतु प्रकार ड्रॉप-डाउन सूचीमधून PostgreSQL निवडा.

काटेकोरपणे सांगायचे तर, मूल्य ::/0 म्हणजे जगभरातून सर्व्हरवर येणार्‍या रहदारीची उपलब्धता, जे प्रमाणिकदृष्ट्या पूर्णपणे सत्य नाही, परंतु उदाहरणाचे विश्लेषण करण्यासाठी, आपण स्वतःला हा दृष्टिकोन वापरण्याची परवानगी देऊ या:

आम्ही ब्राउझर पृष्ठावर परत आलो, जिथे आमच्याकडे "प्रगत सेटिंग्ज कॉन्फिगर करा" उघडले आहे आणि VPC सुरक्षा गट विभागात निवडा -> विद्यमान VPC सुरक्षा गट निवडा -> PostgreSQL:

पुढे, डेटाबेस पर्यायांमध्ये -> डेटाबेस नाव -> नाव सेट करा - habrDB.

आम्ही डीफॉल्टनुसार, बॅकअप अक्षम करणे (बॅकअप धारणा कालावधी - 0 दिवस), मॉनिटरिंग आणि कार्यप्रदर्शन अंतर्दृष्टी वगळता उर्वरित पॅरामीटर्स सोडू शकतो. बटणावर क्लिक करा डेटाबेस तयार करा:

धागा हाताळणारा

अंतिम टप्पा स्पार्क जॉबचा विकास असेल, जो दर दोन सेकंदांनी काफ्काकडून येणाऱ्या नवीन डेटावर प्रक्रिया करेल आणि डेटाबेसमध्ये निकाल प्रविष्ट करेल.

वर नमूद केल्याप्रमाणे, चेकपॉईंट ही स्पार्कस्ट्रीमिंगमधील मुख्य यंत्रणा आहे जी दोष सहनशीलता सुनिश्चित करण्यासाठी कॉन्फिगर केलेली असणे आवश्यक आहे. आम्ही चेकपॉईंट्स वापरू आणि प्रक्रिया अयशस्वी झाल्यास, स्पार्क स्ट्रीमिंग मॉड्यूलला फक्त शेवटच्या चेकपॉईंटवर परत जावे लागेल आणि गमावलेला डेटा पुनर्प्राप्त करण्यासाठी तेथून गणना पुन्हा सुरू करावी लागेल.

दोष-सहिष्णु, विश्वासार्ह फाइल सिस्टम (जसे की HDFS, S3, इ.) वर निर्देशिका सेट करून चेकपॉईंटिंग सक्षम केले जाऊ शकते ज्यामध्ये चेकपॉईंट माहिती संग्रहित केली जाईल. हे वापरून केले जाते, उदाहरणार्थ:

streamingContext.checkpoint(checkpointDirectory)

आमच्या उदाहरणात, आम्ही खालील दृष्टीकोन वापरु, म्हणजे, चेकपॉईंट डायरेक्टरी अस्तित्वात असल्यास, चेकपॉईंट डेटामधून संदर्भ पुन्हा तयार केला जाईल. जर निर्देशिका अस्तित्वात नसेल (म्हणजे प्रथमच कार्यान्वित केली असेल), तर नवीन संदर्भ तयार करण्यासाठी आणि DStreams कॉन्फिगर करण्यासाठी functionToCreateContext कॉल केला जाईल:

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

आम्ही 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")

स्पार्क एसक्यूएल वापरून, आम्ही एक साधे गटबद्ध करतो आणि कन्सोलमध्ये निकाल प्रदर्शित करतो:

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

क्वेरी मजकूर मिळवणे आणि स्पार्क एसक्यूएल द्वारे चालवणे:

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

आणि मग आम्ही परिणामी एकत्रित डेटा AWS RDS मध्ये टेबलमध्ये सेव्ह करतो. डेटाबेस टेबलवर एकत्रित परिणाम जतन करण्यासाठी, आम्ही डेटाफ्रेम ऑब्जेक्टची लेखन पद्धत वापरू:

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 उपयोजित करणे" चरणावर त्यासाठी वापरकर्ता आणि पासवर्ड तयार केला. तुम्ही डेटाबेस सर्व्हर url म्हणून Endpoint चा वापर करावा, जो कनेक्टिव्हिटी आणि सुरक्षा विभागात प्रदर्शित केला जातो:

स्पार्क आणि काफ्का यांना योग्यरित्या जोडण्यासाठी, तुम्ही आर्टिफॅक्ट वापरून smark-submit द्वारे काम चालवावे. spark-streaming-kafka-0-8_2.11. याव्यतिरिक्त, आम्ही PostgreSQL डेटाबेसशी संवाद साधण्यासाठी आर्टिफॅक्ट देखील वापरू; आम्ही ते --packages द्वारे हस्तांतरित करू.

स्क्रिप्टच्या लवचिकतेसाठी, आम्ही इनपुट पॅरामीटर्स म्हणून संदेश सर्व्हरचे नाव आणि ज्या विषयावरून डेटा प्राप्त करू इच्छितो ते देखील समाविष्ट करू.

तर, सिस्टमची कार्यक्षमता लॉन्च करण्याची आणि तपासण्याची वेळ आली आहे:

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 सेकंदांवर सेट करतो:

पुढे, आम्ही टेबलमधील रेकॉर्डची उपस्थिती तपासण्यासाठी डेटाबेसमध्ये एक सोपी क्वेरी करतो व्यवहार_प्रवाह:

निष्कर्ष

हा लेख Apache Kafka आणि PostgreSQL च्या संयोगाने स्पार्क स्ट्रीमिंग वापरून माहितीच्या प्रवाह प्रक्रियेचे उदाहरण पाहतो. विविध स्त्रोतांकडील डेटाच्या वाढीसह, स्ट्रीमिंग आणि रिअल-टाइम ऍप्लिकेशन्स तयार करण्यासाठी स्पार्क स्ट्रीमिंगच्या व्यावहारिक मूल्याचा अतिरेक करणे कठीण आहे.

माझ्या रेपॉजिटरीमध्ये तुम्हाला पूर्ण स्त्रोत कोड येथे सापडेल GitHub.

या लेखावर चर्चा करताना मला आनंद होत आहे, मी तुमच्या टिप्पण्यांसाठी उत्सुक आहे आणि मला सर्व काळजीवाहू वाचकांकडून रचनात्मक टीका करण्याची देखील अपेक्षा आहे.

मी तुम्हाला यश इच्छितो!

PS सुरुवातीला स्थानिक PostgreSQL डेटाबेस वापरण्याची योजना होती, परंतु AWS बद्दलचे माझे प्रेम पाहता, मी डेटाबेस क्लाउडवर हलवण्याचा निर्णय घेतला. या विषयावरील पुढील लेखात, मी AWS मध्ये वर वर्णन केलेली संपूर्ण प्रणाली AWS Kinesis आणि AWS EMR वापरून कशी अंमलात आणायची ते दाखवेन. बातम्यांचे अनुसरण करा!

स्त्रोत: www.habr.com

स्पार्क स्ट्रीमिंगसह अपाचे काफ्का आणि स्ट्रीमिंग डेटा प्रोसेसिंग