अपाचे काफ्का और स्पार्क स्ट्रीमिंग के साथ स्ट्रीमिंग डेटा प्रोसेसिंग

नमस्ते, हबर! आज हम एक ऐसी प्रणाली बनाएंगे जो स्पार्क स्ट्रीमिंग का उपयोग करके अपाचे काफ्का संदेश स्ट्रीम को संसाधित करेगी और प्रसंस्करण परिणामों को एडब्ल्यूएस आरडीएस क्लाउड डेटाबेस में लिखेगी।

आइए कल्पना करें कि एक निश्चित क्रेडिट संस्थान हमें अपनी सभी शाखाओं में आने वाले लेनदेन को "तत्काल" संसाधित करने का कार्य निर्धारित करता है। यह राजकोष के लिए खुली मुद्रा की स्थिति, लेनदेन की सीमा या वित्तीय परिणाम आदि की तुरंत गणना करने के उद्देश्य से किया जा सकता है।

जादू और जादुई मंत्रों के उपयोग के बिना इस मामले को कैसे कार्यान्वित करें - कट के नीचे पढ़ें! जाना!

(छवि स्रोत)

परिचय

बेशक, वास्तविक समय में बड़ी मात्रा में डेटा संसाधित करने से आधुनिक प्रणालियों में उपयोग के पर्याप्त अवसर मिलते हैं। इसके लिए सबसे लोकप्रिय संयोजनों में से एक अपाचे काफ्का और स्पार्क स्ट्रीमिंग का अग्रानुक्रम है, जहां काफ्का आने वाले संदेश पैकेटों की एक धारा बनाता है, और स्पार्क स्ट्रीमिंग इन पैकेटों को एक निश्चित समय अंतराल पर संसाधित करता है।

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

विकसित प्रणाली की वास्तुकला

प्रयुक्त घटक:

• अपाचे काफ्का एक वितरित प्रकाशन-सदस्यता संदेश प्रणाली है। ऑफ़लाइन और ऑनलाइन संदेश उपभोग दोनों के लिए उपयुक्त। डेटा हानि को रोकने के लिए, काफ्का संदेशों को डिस्क पर संग्रहीत किया जाता है और क्लस्टर के भीतर दोहराया जाता है। काफ्का प्रणाली ज़ूकीपर सिंक्रोनाइज़ेशन सेवा के शीर्ष पर बनाई गई है;
• अपाचे स्पार्क स्ट्रीमिंग - स्ट्रीमिंग डेटा को संसाधित करने के लिए स्पार्क घटक। स्पार्क स्ट्रीमिंग मॉड्यूल एक माइक्रो-बैच आर्किटेक्चर का उपयोग करके बनाया गया है, जहां डेटा स्ट्रीम को छोटे डेटा पैकेट के निरंतर अनुक्रम के रूप में व्याख्या किया जाता है। स्पार्क स्ट्रीमिंग विभिन्न स्रोतों से डेटा लेती है और इसे छोटे पैकेजों में जोड़ती है। नियमित अंतराल पर नये पैकेज बनाये जाते हैं। प्रत्येक समय अंतराल की शुरुआत में, एक नया पैकेट बनाया जाता है, और उस अंतराल के दौरान प्राप्त कोई भी डेटा पैकेट में शामिल किया जाता है। अंतराल के अंत में, पैकेट की वृद्धि रुक ​​जाती है। अंतराल का आकार बैच अंतराल नामक पैरामीटर द्वारा निर्धारित किया जाता है;
• अपाचे स्पार्क एसक्यूएल - स्पार्क कार्यात्मक प्रोग्रामिंग के साथ रिलेशनल प्रोसेसिंग को जोड़ती है। संरचित डेटा का मतलब वह डेटा है जिसमें एक स्कीमा होता है, यानी सभी रिकॉर्ड के लिए फ़ील्ड का एक सेट। स्पार्क एसक्यूएल विभिन्न संरचित डेटा स्रोतों से इनपुट का समर्थन करता है और, स्कीमा जानकारी की उपलब्धता के लिए धन्यवाद, यह केवल रिकॉर्ड के आवश्यक फ़ील्ड को कुशलतापूर्वक पुनर्प्राप्त कर सकता है, और डेटाफ़्रेम एपीआई भी प्रदान करता है;
• एडब्ल्यूएस आरडीएस एक अपेक्षाकृत सस्ती क्लाउड-आधारित रिलेशनल डेटाबेस, वेब सेवा है जो सेटअप, संचालन और स्केलिंग को सरल बनाती है, और सीधे अमेज़ॅन द्वारा प्रशासित होती है।

काफ्का सर्वर को स्थापित करना और चलाना

सीधे काफ्का का उपयोग करने से पहले, आपको यह सुनिश्चित करना होगा कि आपके पास जावा है, क्योंकि... 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

काफ्का सर्वर शुरू करने से पहले, आपको ज़ूकीपर सर्वर शुरू करना होगा; हम काफ्का वितरण के साथ आने वाली सहायक स्क्रिप्ट का उपयोग करेंगे:

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

ज़ूकीपर के सफलतापूर्वक प्रारंभ होने के बाद, काफ्का सर्वर को एक अलग टर्मिनल में लॉन्च करें:

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

आइए नव निर्मित विषय के लिए निर्माता और उपभोक्ता के परीक्षण के क्षणों को याद करें। आप संदेश भेजने और प्राप्त करने का परीक्षण कैसे कर सकते हैं, इसके बारे में अधिक विवरण आधिकारिक दस्तावेज़ में लिखा गया है - कुछ संदेश भेजें. खैर, हम काफ्काप्रोड्यूसर एपीआई का उपयोग करके पायथन में एक निर्माता लिखने के लिए आगे बढ़ते हैं।

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

निर्माता यादृच्छिक डेटा उत्पन्न करेगा - प्रति सेकंड 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 और Next बटन पर क्लिक करें:

अगले पृष्ठ पर हमारे डेटाबेस सर्वर की बाहर से पहुंच (सार्वजनिक पहुंच) और पोर्ट उपलब्धता के लिए जिम्मेदार पैरामीटर हैं:

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

हम सुरक्षा समूह के लिए नाम निर्धारित करते हैं - PostgreSQL, एक विवरण, इंगित करें कि इस समूह को किस VPC से संबद्ध किया जाना चाहिए और बनाएं बटन पर क्लिक करें:

नव निर्मित समूह के लिए पोर्ट 5432 के लिए इनबाउंड नियम भरें, जैसा कि नीचे चित्र में दिखाया गया है। आप पोर्ट को मैन्युअल रूप से निर्दिष्ट नहीं कर सकते हैं, लेकिन टाइप ड्रॉप-डाउन सूची से PostgreSQL का चयन करें।

कड़ाई से बोलते हुए, मान ::/0 का मतलब दुनिया भर से सर्वर पर आने वाले ट्रैफ़िक की उपलब्धता है, जो कि पूरी तरह से सच नहीं है, लेकिन उदाहरण का विश्लेषण करने के लिए, आइए हम खुद को इस दृष्टिकोण का उपयोग करने की अनुमति दें:

हम ब्राउज़र पेज पर लौटते हैं, जहां हमने "उन्नत सेटिंग्स कॉन्फ़िगर करें" खोला है और वीपीसी सुरक्षा समूह अनुभाग में चयन करें -> मौजूदा वीपीसी सुरक्षा समूह चुनें -> पोस्टग्रेएसक्यूएल:

इसके बाद, डेटाबेस विकल्पों में -> डेटाबेस नाम -> नाम सेट करें - habrDB.

हम डिफ़ॉल्ट रूप से बैकअप (बैकअप अवधारण अवधि - 0 दिन), निगरानी और प्रदर्शन अंतर्दृष्टि को अक्षम करने के अपवाद के साथ, शेष पैरामीटर छोड़ सकते हैं। बटन पर क्लिक करें डेटाबेस बनाएँ:

धागा संचालक

अंतिम चरण स्पार्क जॉब का विकास होगा, जो हर दो सेकंड में काफ्का से आने वाले नए डेटा को संसाधित करेगा और परिणाम को डेटाबेस में दर्ज करेगा।

जैसा कि ऊपर उल्लेख किया गया है, स्पार्कस्ट्रीमिंग में चेकपॉइंट एक मुख्य तंत्र हैं जिन्हें गलती सहनशीलता सुनिश्चित करने के लिए कॉन्फ़िगर किया जाना चाहिए। हम चेकपॉइंट का उपयोग करेंगे और, यदि प्रक्रिया विफल हो जाती है, तो स्पार्क स्ट्रीमिंग मॉड्यूल को केवल अंतिम चेकपॉइंट पर लौटने और खोए हुए डेटा को पुनर्प्राप्त करने के लिए गणना फिर से शुरू करने की आवश्यकता होगी।

चेकपॉइंटिंग को दोष-सहिष्णु, विश्वसनीय फ़ाइल सिस्टम (जैसे HDFS, S3, आदि) पर एक निर्देशिका सेट करके सक्षम किया जा सकता है जिसमें चेकपॉइंट जानकारी संग्रहीत की जाएगी। उदाहरण के लिए, इसका उपयोग करके किया जाता है:

streamingContext.checkpoint(checkpointDirectory)

हमारे उदाहरण में, हम निम्नलिखित दृष्टिकोण का उपयोग करेंगे, अर्थात्, यदि चेकपॉइंटडायरेक्टरी मौजूद है, तो संदर्भ को चेकपॉइंट डेटा से फिर से बनाया जाएगा। यदि निर्देशिका मौजूद नहीं है (यानी पहली बार निष्पादित की गई है), तो नया संदर्भ बनाने और DStreams को कॉन्फ़िगर करने के लिए functionToCreateContext को कॉल किया जाता है:

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

हम KafkaUtils लाइब्रेरी की createDirectStream पद्धति का उपयोग करके "लेन-देन" विषय से जुड़ने के लिए एक DirectStream ऑब्जेक्ट बनाते हैं:

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 के माध्यम से चलाना:

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 से कनेक्शन स्थापित करने के बारे में कुछ शब्द। हमने "एडब्ल्यूएस पोस्टग्रेएसक्यूएल परिनियोजन" चरण पर इसके लिए उपयोगकर्ता और पासवर्ड बनाया। आपको डेटाबेस सर्वर यूआरएल के रूप में एंडपॉइंट का उपयोग करना चाहिए, जो कनेक्टिविटी और सुरक्षा अनुभाग में प्रदर्शित होता है:

स्पार्क और काफ्का को सही ढंग से कनेक्ट करने के लिए, आपको आर्टिफैक्ट का उपयोग करके स्मार्क-सबमिट के माध्यम से कार्य चलाना चाहिए स्पार्क-स्ट्रीमिंग-काफ्का-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.

मुझे इस लेख पर चर्चा करने में खुशी हो रही है, मैं आपकी टिप्पणियों की प्रतीक्षा कर रहा हूं, और मैं सभी परवाह करने वाले पाठकों से रचनात्मक आलोचना की भी आशा करता हूं।

मैं आपको सफलता की कामना करता हूं!

भज। शुरुआत में स्थानीय PostgreSQL डेटाबेस का उपयोग करने की योजना बनाई गई थी, लेकिन AWS के प्रति मेरे प्यार को देखते हुए, मैंने डेटाबेस को क्लाउड पर स्थानांतरित करने का निर्णय लिया। इस विषय पर अगले लेख में, मैं दिखाऊंगा कि एडब्ल्यूएस किनेसिस और एडब्ल्यूएस ईएमआर का उपयोग करके एडब्ल्यूएस में ऊपर वर्णित संपूर्ण प्रणाली को कैसे लागू किया जाए। समाचार का पालन करें!

स्रोत: www.habr.com