સ્પાર્ક સ્ટ્રીમિંગ સાથે અપાચે કાફકા અને સ્ટ્રીમિંગ ડેટા પ્રોસેસિંગ

હેલો, હેબ્ર! આજે આપણે એવી સિસ્ટમ બનાવીશું જે સ્પાર્ક સ્ટ્રીમિંગનો ઉપયોગ કરીને Apache Kafka મેસેજ સ્ટ્રીમ પર પ્રક્રિયા કરશે અને પ્રોસેસિંગ પરિણામોને AWS RDS ક્લાઉડ ડેટાબેઝમાં લખશે.

ચાલો કલ્પના કરીએ કે કોઈ ચોક્કસ ધિરાણ સંસ્થા અમને તેની તમામ શાખાઓમાં "ફ્લાય પર" આવતા વ્યવહારોની પ્રક્રિયા કરવાનું કાર્ય સુયોજિત કરે છે. ટ્રેઝરી માટે ખુલ્લા ચલણની સ્થિતિ, વ્યવહારો માટે મર્યાદા અથવા નાણાકીય પરિણામો વગેરેની તાત્કાલિક ગણતરી કરવાના હેતુથી આ કરી શકાય છે.

જાદુ અને જાદુઈ જોડણીના ઉપયોગ વિના આ કેસને કેવી રીતે અમલમાં મૂકવો - કટ હેઠળ વાંચો! જાઓ!

(છબી સ્ત્રોત)

પરિચય

અલબત્ત, રીઅલ ટાઇમમાં મોટી માત્રામાં ડેટાની પ્રક્રિયા આધુનિક સિસ્ટમમાં ઉપયોગ માટે પૂરતી તકો પૂરી પાડે છે. આ માટેના સૌથી લોકપ્રિય સંયોજનોમાંનું એક અપાચે કાફકા અને સ્પાર્ક સ્ટ્રીમિંગનું ટેન્ડમ છે, જ્યાં કાફકા ઇનકમિંગ મેસેજ પેકેટોનો સ્ટ્રીમ બનાવે છે, અને સ્પાર્ક સ્ટ્રીમિંગ આ પેકેટોને આપેલ સમય અંતરાલ પર પ્રક્રિયા કરે છે.

એપ્લિકેશનની ખામી સહિષ્ણુતા વધારવા માટે, અમે ચેકપોઇન્ટ્સનો ઉપયોગ કરીશું. આ મિકેનિઝમ સાથે, જ્યારે સ્પાર્ક સ્ટ્રીમિંગ એન્જિનને ખોવાયેલો ડેટા પુનઃપ્રાપ્ત કરવાની જરૂર પડે છે, ત્યારે તેને માત્ર છેલ્લા ચેકપોઇન્ટ પર પાછા જવાની અને ત્યાંથી ગણતરીઓ ફરી શરૂ કરવાની જરૂર છે.

વિકસિત સિસ્ટમનું આર્કિટેક્ચર

વપરાયેલ ઘટકો:

• અપાચે કાફકા વિતરિત પબ્લિશ-સબ્સ્ક્રાઇબ મેસેજિંગ સિસ્ટમ છે. ઓફલાઈન અને ઓનલાઈન બંને સંદેશ વપરાશ માટે યોગ્ય. ડેટા નુકશાન અટકાવવા માટે, કાફકા સંદેશાઓ ડિસ્ક પર સંગ્રહિત થાય છે અને ક્લસ્ટરની અંદર નકલ કરવામાં આવે છે. કાફકા સિસ્ટમ ઝૂકીપર સિંક્રોનાઇઝેશન સેવાની ટોચ પર બનેલી છે;
• અપાચે સ્પાર્ક સ્ટ્રીમિંગ - સ્ટ્રીમિંગ ડેટાની પ્રક્રિયા કરવા માટે સ્પાર્ક ઘટક. સ્પાર્ક સ્ટ્રીમિંગ મોડ્યુલ માઇક્રો-બેચ આર્કિટેક્ચરનો ઉપયોગ કરીને બનાવવામાં આવ્યું છે, જ્યાં ડેટા સ્ટ્રીમને નાના ડેટા પેકેટોના સતત ક્રમ તરીકે અર્થઘટન કરવામાં આવે છે. સ્પાર્ક સ્ટ્રીમિંગ વિવિધ સ્ત્રોતોમાંથી ડેટા લે છે અને તેને નાના પેકેજોમાં જોડે છે. નવા પેકેજો નિયમિત સમયાંતરે બનાવવામાં આવે છે. દરેક સમય અંતરાલની શરૂઆતમાં, એક નવું પેકેટ બનાવવામાં આવે છે, અને તે અંતરાલ દરમિયાન પ્રાપ્ત થયેલ કોઈપણ ડેટા પેકેટમાં સમાવવામાં આવે છે. અંતરાલના અંતે, પેકેટ વૃદ્ધિ અટકે છે. અંતરાલનું કદ બેચ અંતરાલ તરીકે ઓળખાતા પરિમાણ દ્વારા નક્કી કરવામાં આવે છે;
• અપાચે સ્પાર્ક એસક્યુએલ - સ્પાર્ક ફંક્શનલ પ્રોગ્રામિંગ સાથે રિલેશનલ પ્રોસેસિંગને જોડે છે. સ્ટ્રક્ચર્ડ ડેટા એટલે કે સ્કીમા ધરાવતો ડેટા, એટલે કે તમામ રેકોર્ડ્સ માટે ફીલ્ડનો એક સેટ. સ્પાર્ક એસક્યુએલ વિવિધ માળખાગત ડેટા સ્ત્રોતોમાંથી ઇનપુટને સમર્થન આપે છે અને, સ્કીમા માહિતીની ઉપલબ્ધતાને આભારી છે, તે માત્ર રેકોર્ડ્સના જરૂરી ક્ષેત્રોને અસરકારક રીતે પુનઃપ્રાપ્ત કરી શકે છે, અને ડેટાફ્રેમ API પણ પ્રદાન કરે છે;
• AWS RDS પ્રમાણમાં સસ્તું ક્લાઉડ-આધારિત રિલેશનલ ડેટાબેઝ છે, વેબ સેવા કે જે સેટઅપ, ઓપરેશન અને સ્કેલિંગને સરળ બનાવે છે, અને સીધા એમેઝોન દ્વારા સંચાલિત થાય છે.

કાફકા સર્વરને ઇન્સ્ટોલ અને ચલાવી રહ્યું છે

કાફકાનો સીધો ઉપયોગ કરતા પહેલા, તમારે ખાતરી કરવાની જરૂર છે કે તમારી પાસે જાવા છે, કારણ કે... 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

ચાલો નવા બનાવેલા વિષય માટે નિર્માતા અને ઉપભોક્તાનું પરીક્ષણ કરવાની ક્ષણો ચૂકી જઈએ. તમે સંદેશા મોકલવા અને પ્રાપ્ત કરવાનું પરીક્ષણ કેવી રીતે કરી શકો તે વિશે વધુ વિગતો સત્તાવાર દસ્તાવેજોમાં લખેલી છે - કેટલાક સંદેશાઓ મોકલો. સારું, અમે કાફકાપ્રોડ્યુસર 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()

સ્ક્રિપ્ટ ચલાવતી વખતે, અમને ટર્મિનલમાં નીચેના સંદેશાઓ પ્રાપ્ત થાય છે:

આનો અર્થ એ છે કે બધું અમે ઇચ્છીએ છીએ તે પ્રમાણે કાર્ય કરે છે - નિર્માતા અમને જરૂરી વિષય પર સંદેશા જનરેટ કરે છે અને મોકલે છે.
આગળનું પગલું સ્પાર્ક ઇન્સ્ટોલ કરવાનું છે અને આ સંદેશ સ્ટ્રીમ પર પ્રક્રિયા કરવાનું છે.

અપાચે સ્પાર્ક ઇન્સ્ટોલ કરી રહ્યું છે

અપાચે સ્પાર્ક એક સાર્વત્રિક અને ઉચ્ચ-પ્રદર્શન ક્લસ્ટર કમ્પ્યુટિંગ પ્લેટફોર્મ છે.

સ્પાર્ક મેપરેડ્યુસ મોડલના લોકપ્રિય અમલીકરણો કરતાં વધુ સારી કામગીરી કરે છે જ્યારે ઇન્ટરેક્ટિવ ક્વેરીઝ અને સ્ટ્રીમ પ્રોસેસિંગ સહિત ગણતરીના પ્રકારોની વિશાળ શ્રેણીને સમર્થન આપે છે. મોટી માત્રામાં ડેટાની પ્રક્રિયા કરતી વખતે ઝડપ મહત્વની ભૂમિકા ભજવે છે, કારણ કે તે ઝડપ છે જે તમને મિનિટો અથવા કલાકો રાહ જોયા વિના ઇન્ટરેક્ટિવ રીતે કામ કરવાની મંજૂરી આપે છે. સ્પાર્કની સૌથી મોટી શક્તિઓમાંથી એક જે તેને આટલી ઝડપી બનાવે છે તે તેની ઇન-મેમરી ગણતરીઓ કરવાની ક્ષમતા છે.

આ ફ્રેમવર્ક Scala માં લખાયેલ છે, તેથી તમારે તેને પહેલા ઇન્સ્ટોલ કરવાની જરૂર છે:

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 ડેશબોર્ડ -> સુરક્ષા જૂથો -> સુરક્ષા જૂથ વિભાગ બનાવો:

અમે સુરક્ષા જૂથ માટે નામ સેટ કર્યું છે - PostgreSQL, એક વર્ણન, જે દર્શાવે છે કે આ જૂથ કયા VPC સાથે સંકળાયેલું હોવું જોઈએ અને બનાવો બટનને ક્લિક કરો:

નીચેના ચિત્રમાં બતાવ્યા પ્રમાણે નવા બનાવેલા જૂથ માટે પોર્ટ 5432 માટેના ઇનબાઉન્ડ નિયમો ભરો. તમે પોર્ટને મેન્યુઅલી સ્પષ્ટ કરી શકતા નથી, પરંતુ પ્રકાર ડ્રોપ-ડાઉન સૂચિમાંથી PostgreSQL પસંદ કરો.

કડક શબ્દોમાં કહીએ તો, મૂલ્ય ::/0 નો અર્થ સમગ્ર વિશ્વમાંથી સર્વર પર આવતા ટ્રાફિકની ઉપલબ્ધતા છે, જે પ્રમાણભૂત રીતે સંપૂર્ણ રીતે સાચું નથી, પરંતુ ઉદાહરણનું વિશ્લેષણ કરવા માટે, ચાલો આપણે આ અભિગમનો ઉપયોગ કરવાની મંજૂરી આપીએ:

અમે બ્રાઉઝર પૃષ્ઠ પર પાછા ફરીએ છીએ, જ્યાં અમારી પાસે "અદ્યતન સેટિંગ્સ ગોઠવો" ખુલ્લું છે અને VPC સુરક્ષા જૂથો વિભાગમાં પસંદ કરો -> હાલના VPC સુરક્ષા જૂથો પસંદ કરો -> PostgreSQL:

આગળ, ડેટાબેઝ વિકલ્પોમાં -> ડેટાબેઝ નામ -> નામ સેટ કરો - habrDB.

અમે ડિફૉલ્ટ રૂપે, બેકઅપ (બેકઅપ રીટેન્શન પીરિયડ - 0 દિવસ), મોનિટરિંગ અને પ્રદર્શન આંતરદૃષ્ટિને અક્ષમ કરવાના અપવાદ સિવાય બાકીના પરિમાણો છોડી શકીએ છીએ. બટન પર ક્લિક કરો ડેટાબેઝ બનાવો:

થ્રેડ હેન્ડલર

અંતિમ તબક્કો સ્પાર્ક જોબનો વિકાસ હશે, જે દર બે સેકન્ડે કાફકા તરફથી આવતા નવા ડેટા પર પ્રક્રિયા કરશે અને પરિણામ ડેટાબેઝમાં દાખલ કરશે.

ઉપર નોંધ્યું છે તેમ, ચેકપોઇન્ટ્સ સ્પાર્કસ્ટ્રીમિંગમાં એક મુખ્ય પદ્ધતિ છે જે દોષ સહિષ્ણુતાને સુનિશ્ચિત કરવા માટે ગોઠવેલ હોવી આવશ્યક છે. અમે ચેકપોઇન્ટ્સનો ઉપયોગ કરીશું અને, જો પ્રક્રિયા નિષ્ફળ જાય, તો સ્પાર્ક સ્ટ્રીમિંગ મોડ્યુલને ફક્ત છેલ્લા ચેકપોઇન્ટ પર પાછા ફરવાની અને ખોવાયેલ ડેટાને પુનઃપ્રાપ્ત કરવા માટે તેમાંથી ગણતરીઓ ફરી શરૂ કરવાની જરૂર પડશે.

ચેકપૉઇન્ટિંગને ખામી-સહિષ્ણુ, વિશ્વસનીય ફાઇલ સિસ્ટમ (જેમ કે HDFS, S3, વગેરે) પર ડિરેક્ટરી સેટ કરીને સક્ષમ કરી શકાય છે જેમાં ચેકપોઇન્ટ માહિતી સંગ્રહિત કરવામાં આવશે. આનો ઉપયોગ કરીને કરવામાં આવે છે, ઉદાહરણ તરીકે:

streamingContext.checkpoint(checkpointDirectory)

અમારા ઉદાહરણમાં, અમે નીચેના અભિગમનો ઉપયોગ કરીશું, એટલે કે, જો checkpointDirectory અસ્તિત્વમાં છે, તો પછી સંદર્ભ ચેકપોઇન્ટ ડેટામાંથી ફરીથી બનાવવામાં આવશે. જો ડિરેક્ટરી અસ્તિત્વમાં ન હોય (એટલે ​​​​કે પ્રથમ વખત એક્ઝિક્યુટ કરવામાં આવી હોય), તો ફંક્શનToCreateContext ને નવો સંદર્ભ બનાવવા અને DStreams ને ગોઠવવા માટે કહેવામાં આવે છે:

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 તરીકે એન્ડપોઇન્ટનો ઉપયોગ કરવો જોઈએ, જે કનેક્ટિવિટી અને સુરક્ષા વિભાગમાં પ્રદર્શિત થાય છે:

સ્પાર્ક અને કાફકાને યોગ્ય રીતે જોડવા માટે, તમારે આર્ટિફેક્ટનો ઉપયોગ કરીને smark-સબમિટ દ્વારા જોબ ચલાવવી જોઈએ. સ્પાર્ક-સ્ટ્રીમિંગ-કાફ્કા-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 સેકન્ડ પર સેટ કરીએ છીએ:

આગળ, અમે કોષ્ટકમાં રેકોર્ડની હાજરી તપાસવા માટે ડેટાબેઝમાં એક સરળ ક્વેરી કરીએ છીએ વ્યવહાર_પ્રવાહ:

નિષ્કર્ષ

આ લેખ અપાચે કાફકા અને પોસ્ટગ્રેએસક્યુએલ સાથે જોડાણમાં સ્પાર્ક સ્ટ્રીમિંગનો ઉપયોગ કરીને માહિતીના સ્ટ્રીમ પ્રોસેસિંગના ઉદાહરણ પર ધ્યાન આપે છે. વિવિધ સ્ત્રોતોમાંથી ડેટાની વૃદ્ધિ સાથે, સ્ટ્રીમિંગ અને રીઅલ-ટાઇમ એપ્લિકેશન્સ બનાવવા માટે સ્પાર્ક સ્ટ્રીમિંગના વ્યવહારિક મૂલ્યને વધુ પડતો અંદાજ કાઢવો મુશ્કેલ છે.

તમે મારા રીપોઝીટરીમાં સંપૂર્ણ સ્ત્રોત કોડ અહીંથી મેળવી શકો છો GitHub.

હું આ લેખની ચર્ચા કરવામાં ખુશ છું, હું તમારી ટિપ્પણીઓની રાહ જોઉં છું, અને હું બધા કાળજી રાખનારા વાચકો તરફથી રચનાત્મક ટીકાની પણ આશા રાખું છું.

હું તમને સફળતાની ઇચ્છા કરું છું!

ગીત. શરૂઆતમાં સ્થાનિક PostgreSQL ડેટાબેઝનો ઉપયોગ કરવાનું આયોજન કરવામાં આવ્યું હતું, પરંતુ AWS પ્રત્યેના મારા પ્રેમને જોતાં, મેં ડેટાબેઝને ક્લાઉડ પર ખસેડવાનું નક્કી કર્યું. આ વિષય પરના આગલા લેખમાં, હું AWS માં ઉપર વર્ણવેલ સમગ્ર સિસ્ટમને AWS Kinesis અને AWS EMR નો ઉપયોગ કરીને કેવી રીતે અમલમાં મૂકવી તે બતાવીશ. સમાચાર અનુસરો!

સોર્સ: www.habr.com