അപ്പാച്ചെ കാഫ്കയും സ്പാർക്ക് സ്ട്രീമിംഗിനൊപ്പം സ്ട്രീമിംഗ് ഡാറ്റ പ്രോസസ്സിംഗും

ഹലോ, ഹബ്ർ! ഇന്ന് ഞങ്ങൾ സ്പാർക്ക് സ്ട്രീമിംഗ് ഉപയോഗിച്ച് അപ്പാച്ചെ കാഫ്ക സന്ദേശ സ്ട്രീമുകൾ പ്രോസസ്സ് ചെയ്യുന്ന ഒരു സിസ്റ്റം നിർമ്മിക്കുകയും പ്രോസസ്സിംഗ് ഫലങ്ങൾ AWS RDS ക്ലൗഡ് ഡാറ്റാബേസിൽ എഴുതുകയും ചെയ്യും.

ഒരു പ്രത്യേക ക്രെഡിറ്റ് സ്ഥാപനം അതിന്റെ എല്ലാ ശാഖകളിലുമുള്ള ഇൻകമിംഗ് ഇടപാടുകൾ "ഈച്ചയിൽ" പ്രോസസ്സ് ചെയ്യുന്നതിനുള്ള ചുമതല ഞങ്ങളെ സജ്ജമാക്കുന്നുവെന്ന് സങ്കൽപ്പിക്കുക. ട്രഷറിയുടെ ഓപ്പൺ കറൻസി സ്ഥാനം, ഇടപാടുകൾക്കുള്ള പരിധികൾ അല്ലെങ്കിൽ സാമ്പത്തിക ഫലങ്ങൾ മുതലായവ പെട്ടെന്ന് കണക്കുകൂട്ടാൻ ഇത് ചെയ്യാവുന്നതാണ്.

മാന്ത്രികവും മാന്ത്രിക മന്ത്രങ്ങളും ഉപയോഗിക്കാതെ ഈ കേസ് എങ്ങനെ നടപ്പിലാക്കാം - കട്ടിന് കീഴിൽ വായിക്കുക! പോകൂ!

(ചിത്രത്തിന്റെ ഉറവിടം)

ആമുഖം

തീർച്ചയായും, വലിയ അളവിലുള്ള ഡാറ്റ തത്സമയം പ്രോസസ്സ് ചെയ്യുന്നത് ആധുനിക സിസ്റ്റങ്ങളിൽ ഉപയോഗിക്കുന്നതിന് ധാരാളം അവസരങ്ങൾ നൽകുന്നു. ഇതിനുള്ള ഏറ്റവും ജനപ്രിയമായ കോമ്പിനേഷനുകളിലൊന്നാണ് അപ്പാച്ചെ കാഫ്കയുടെയും സ്പാർക്ക് സ്ട്രീമിംഗിന്റെയും സംയോജനം, അവിടെ കാഫ്ക ഇൻകമിംഗ് സന്ദേശ പാക്കറ്റുകളുടെ ഒരു സ്ട്രീം സൃഷ്ടിക്കുന്നു, കൂടാതെ സ്പാർക്ക് സ്ട്രീമിംഗ് ഈ പാക്കറ്റുകളെ ഒരു നിശ്ചിത സമയ ഇടവേളയിൽ പ്രോസസ്സ് ചെയ്യുന്നു.

ആപ്ലിക്കേഷന്റെ തെറ്റ് സഹിഷ്ണുത വർദ്ധിപ്പിക്കുന്നതിന്, ഞങ്ങൾ ചെക്ക് പോയിന്റുകൾ ഉപയോഗിക്കും. ഈ സംവിധാനം ഉപയോഗിച്ച്, സ്പാർക്ക് സ്ട്രീമിംഗ് എഞ്ചിന് നഷ്ടപ്പെട്ട ഡാറ്റ വീണ്ടെടുക്കേണ്ടിവരുമ്പോൾ, അവസാന ചെക്ക് പോയിന്റിലേക്ക് തിരികെ പോയി അവിടെ നിന്ന് കണക്കുകൂട്ടലുകൾ പുനരാരംഭിച്ചാൽ മതിയാകും.

വികസിത സംവിധാനത്തിന്റെ വാസ്തുവിദ്യ

ഉപയോഗിച്ച ഘടകങ്ങൾ:

• അപ്പാച്ച കാഫ്ക വിതരണം ചെയ്ത പ്രസിദ്ധീകരിക്കൽ-സബ്‌സ്‌ക്രൈബ് സന്ദേശമയയ്‌ക്കൽ സംവിധാനമാണ്. ഓഫ്‌ലൈനും ഓൺലൈൻ സന്ദേശ ഉപയോഗത്തിനും അനുയോജ്യം. ഡാറ്റ നഷ്‌ടമാകാതിരിക്കാൻ, കാഫ്ക സന്ദേശങ്ങൾ ഡിസ്കിൽ സംഭരിക്കുകയും ക്ലസ്റ്ററിനുള്ളിൽ പകർത്തുകയും ചെയ്യുന്നു. ZooKeeper സിൻക്രൊണൈസേഷൻ സേവനത്തിന് മുകളിലാണ് കാഫ്ക സംവിധാനം നിർമ്മിച്ചിരിക്കുന്നത്;
• അപ്പാച്ചെ സ്പാർക്ക് സ്ട്രീമിംഗ് - സ്ട്രീമിംഗ് ഡാറ്റ പ്രോസസ്സ് ചെയ്യുന്നതിനുള്ള സ്പാർക്ക് ഘടകം. സ്പാർക്ക് സ്ട്രീമിംഗ് മൊഡ്യൂൾ ഒരു മൈക്രോ ബാച്ച് ആർക്കിടെക്ചർ ഉപയോഗിച്ചാണ് നിർമ്മിച്ചിരിക്കുന്നത്, ഇവിടെ ഡാറ്റ സ്ട്രീം ചെറിയ ഡാറ്റ പാക്കറ്റുകളുടെ തുടർച്ചയായ ശ്രേണിയായി വ്യാഖ്യാനിക്കപ്പെടുന്നു. സ്പാർക്ക് സ്ട്രീമിംഗ് വ്യത്യസ്‌ത സ്രോതസ്സുകളിൽ നിന്ന് ഡാറ്റ എടുത്ത് ചെറിയ പാക്കേജുകളായി സംയോജിപ്പിക്കുന്നു. കൃത്യമായ ഇടവേളകളിൽ പുതിയ പാക്കേജുകൾ സൃഷ്ടിക്കപ്പെടുന്നു. ഓരോ സമയ ഇടവേളയുടെയും തുടക്കത്തിൽ, ഒരു പുതിയ പാക്കറ്റ് സൃഷ്ടിക്കപ്പെടുന്നു, കൂടാതെ ആ ഇടവേളയിൽ ലഭിക്കുന്ന ഏത് ഡാറ്റയും പാക്കറ്റിൽ ഉൾപ്പെടുത്തും. ഇടവേളയുടെ അവസാനം, പാക്കറ്റ് വളർച്ച നിർത്തുന്നു. ഇടവേളയുടെ വലുപ്പം നിർണ്ണയിക്കുന്നത് ബാച്ച് ഇടവേള എന്ന് വിളിക്കുന്ന ഒരു പരാമീറ്ററാണ്;
• അപ്പാച്ചെ സ്പാർക്ക് SQL - റിലേഷണൽ പ്രോസസ്സിംഗും സ്പാർക്ക് ഫംഗ്ഷണൽ പ്രോഗ്രാമിംഗും സംയോജിപ്പിക്കുന്നു. ഘടനാപരമായ ഡാറ്റ എന്നാൽ ഒരു സ്കീമ ഉള്ള ഡാറ്റ എന്നാണ് അർത്ഥമാക്കുന്നത്, അതായത് എല്ലാ റെക്കോർഡുകൾക്കുമായി ഒരൊറ്റ സെറ്റ് ഫീൽഡുകൾ. വിവിധ ഘടനാപരമായ ഡാറ്റാ ഉറവിടങ്ങളിൽ നിന്നുള്ള ഇൻപുട്ടിനെ Spark SQL പിന്തുണയ്ക്കുന്നു, കൂടാതെ സ്കീമ വിവരങ്ങളുടെ ലഭ്യതയ്ക്ക് നന്ദി, അതിന് ആവശ്യമായ റെക്കോർഡുകളുടെ ഫീൽഡുകൾ മാത്രം കാര്യക്ഷമമായി വീണ്ടെടുക്കാൻ കഴിയും, കൂടാതെ DataFrame 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

പുതുതായി സൃഷ്ടിച്ച വിഷയത്തിനായി നിർമ്മാതാവിനെയും ഉപഭോക്താവിനെയും പരീക്ഷിക്കുന്ന നിമിഷങ്ങൾ നമുക്ക് നഷ്ടപ്പെടുത്താം. സന്ദേശങ്ങൾ അയയ്ക്കുന്നതും സ്വീകരിക്കുന്നതും എങ്ങനെ പരിശോധിക്കാം എന്നതിനെക്കുറിച്ചുള്ള കൂടുതൽ വിശദാംശങ്ങൾ ഔദ്യോഗിക ഡോക്യുമെന്റേഷനിൽ എഴുതിയിരിക്കുന്നു - കുറച്ച് സന്ദേശങ്ങൾ അയയ്ക്കുക. ശരി, കാഫ്ക പ്രൊഡ്യൂസർ എപിഐ ഉപയോഗിച്ച് പൈത്തണിൽ ഒരു പ്രൊഡ്യൂസർ എഴുതുന്നതിലേക്ക് ഞങ്ങൾ നീങ്ങുന്നു.

നിർമ്മാതാവിന്റെ എഴുത്ത്

നിർമ്മാതാവ് ക്രമരഹിതമായ ഡാറ്റ സൃഷ്ടിക്കും - ഓരോ സെക്കൻഡിലും 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()

സ്ക്രിപ്റ്റ് പ്രവർത്തിപ്പിക്കുമ്പോൾ, ടെർമിനലിൽ ഞങ്ങൾക്ക് ഇനിപ്പറയുന്ന സന്ദേശങ്ങൾ ലഭിക്കും:

ഇതിനർത്ഥം എല്ലാം നമ്മൾ ആഗ്രഹിച്ചതുപോലെ പ്രവർത്തിക്കുന്നു എന്നാണ് - നിർമ്മാതാവ് നമുക്ക് ആവശ്യമുള്ള വിഷയത്തിലേക്ക് സന്ദേശങ്ങൾ സൃഷ്ടിക്കുകയും അയയ്ക്കുകയും ചെയ്യുന്നു.
സ്പാർക്ക് ഇൻസ്റ്റാൾ ചെയ്ത് ഈ സന്ദേശ സ്ട്രീം പ്രോസസ്സ് ചെയ്യുക എന്നതാണ് അടുത്ത ഘട്ടം.

അപ്പാച്ചെ സ്പാർക്ക് ഇൻസ്റ്റാൾ ചെയ്യുന്നു

അപ്പാച്ചെ സ്പാർക്ക് സാർവത്രികവും ഉയർന്ന പ്രകടനമുള്ളതുമായ ക്ലസ്റ്റർ കമ്പ്യൂട്ടിംഗ് പ്ലാറ്റ്‌ഫോമാണ്.

ഇന്ററാക്ടീവ് ക്വറികളും സ്ട്രീം പ്രോസസ്സിംഗും ഉൾപ്പെടെ വിശാലമായ കമ്പ്യൂട്ടേഷൻ തരങ്ങളെ പിന്തുണയ്‌ക്കുമ്പോൾ MapReduce മോഡലിന്റെ ജനപ്രിയ നിർവ്വഹണങ്ങളേക്കാൾ മികച്ച പ്രകടനം Spark നടത്തുന്നു. വലിയ അളവിലുള്ള ഡാറ്റ പ്രോസസ്സ് ചെയ്യുമ്പോൾ വേഗത ഒരു പ്രധാന പങ്ക് വഹിക്കുന്നു, കാരണം ഇത് മിനിറ്റുകളോ മണിക്കൂറുകളോ കാത്തിരിക്കാതെ സംവേദനാത്മകമായി പ്രവർത്തിക്കാൻ നിങ്ങളെ അനുവദിക്കുന്ന വേഗതയാണ്. സ്പാർക്കിന്റെ ഏറ്റവും വലിയ ശക്തികളിലൊന്ന്, മെമ്മറിയിൽ കണക്കുകൂട്ടലുകൾ നടത്താനുള്ള കഴിവാണ്.

ഈ ചട്ടക്കൂട് സ്കാലയിൽ എഴുതിയിരിക്കുന്നു, അതിനാൽ നിങ്ങൾ ആദ്യം ഇത് ഇൻസ്റ്റാൾ ചെയ്യേണ്ടതുണ്ട്:

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, പ്രധാന ഉപയോക്താവ്: ഹബ്ർ, password: habr12345 അടുത്ത ബട്ടണിൽ ക്ലിക്ക് ചെയ്യുക:

അടുത്ത പേജിൽ പുറത്തുനിന്നുള്ള ഞങ്ങളുടെ ഡാറ്റാബേസ് സെർവറിന്റെ പ്രവേശനക്ഷമതയ്ക്കും (പൊതു പ്രവേശനക്ഷമത) പോർട്ട് ലഭ്യതയ്ക്കും ഉത്തരവാദിത്തമുള്ള പാരാമീറ്ററുകൾ ഉണ്ട്:

VPC സെക്യൂരിറ്റി ഗ്രൂപ്പിനായി നമുക്ക് ഒരു പുതിയ ക്രമീകരണം സൃഷ്ടിക്കാം, അത് പോർട്ട് 5432 (PostgreSQL) വഴി ഞങ്ങളുടെ ഡാറ്റാബേസ് സെർവറിലേക്ക് ബാഹ്യ ആക്സസ് അനുവദിക്കും.
VPC ഡാഷ്‌ബോർഡ് -> സെക്യൂരിറ്റി ഗ്രൂപ്പുകൾ -> സെക്യൂരിറ്റി ഗ്രൂപ്പ് സെക്ഷനിലേക്ക് ഒരു പ്രത്യേക ബ്രൗസർ വിൻഡോയിലെ AWS കൺസോളിലേക്ക് പോകാം:

സെക്യൂരിറ്റി ഗ്രൂപ്പിനായി ഞങ്ങൾ പേര് സജ്ജീകരിച്ചു - PostgreSQL, ഒരു വിവരണം, ഈ ഗ്രൂപ്പ് ഏത് VPC യുമായി ബന്ധപ്പെടുത്തണമെന്ന് സൂചിപ്പിച്ച് സൃഷ്ടിക്കുക ബട്ടൺ ക്ലിക്കുചെയ്യുക:

ചുവടെയുള്ള ചിത്രത്തിൽ കാണിച്ചിരിക്കുന്നതുപോലെ, പുതുതായി സൃഷ്ടിച്ച ഗ്രൂപ്പിനായി പോർട്ട് 5432-നുള്ള ഇൻബൗണ്ട് നിയമങ്ങൾ പൂരിപ്പിക്കുക. നിങ്ങൾക്ക് പോർട്ട് സ്വമേധയാ വ്യക്തമാക്കാൻ കഴിയില്ല, പക്ഷേ ടൈപ്പ് ഡ്രോപ്പ്-ഡൗൺ ലിസ്റ്റിൽ നിന്ന് PostgreSQL തിരഞ്ഞെടുക്കുക.

കൃത്യമായി പറഞ്ഞാൽ, മൂല്യം ::/0 എന്നാൽ ലോകമെമ്പാടുമുള്ള സെർവറിലേക്കുള്ള ഇൻകമിംഗ് ട്രാഫിക്കിന്റെ ലഭ്യത അർത്ഥമാക്കുന്നു, ഇത് കാനോനികമായി പൂർണ്ണമായും ശരിയല്ല, പക്ഷേ ഉദാഹരണം വിശകലനം ചെയ്യാൻ, ഈ സമീപനം ഉപയോഗിക്കാൻ നമുക്ക് സ്വയം അനുവദിക്കാം:

ഞങ്ങൾ ബ്രൗസർ പേജിലേക്ക് മടങ്ങുന്നു, അവിടെ "വിപുലമായ ക്രമീകരണങ്ങൾ കോൺഫിഗർ ചെയ്യുക" തുറന്ന് VPC സെക്യൂരിറ്റി ഗ്രൂപ്പുകളുടെ വിഭാഗത്തിൽ തിരഞ്ഞെടുക്കുക -> നിലവിലുള്ള VPC സുരക്ഷാ ഗ്രൂപ്പുകൾ തിരഞ്ഞെടുക്കുക -> PostgreSQL:

അടുത്തതായി, ഡാറ്റാബേസ് ഓപ്ഷനുകളിൽ -> ഡാറ്റാബേസ് നാമം -> പേര് സജ്ജമാക്കുക - habrDB.

ബാക്കപ്പ് പ്രവർത്തനരഹിതമാക്കുന്നത് (ബാക്കപ്പ് നിലനിർത്തൽ കാലയളവ് - 0 ദിവസം), മോണിറ്ററിംഗ്, പ്രകടന സ്ഥിതിവിവരക്കണക്കുകൾ എന്നിവ ഒഴികെയുള്ള ശേഷിക്കുന്ന പാരാമീറ്ററുകൾ നമുക്ക് ഡിഫോൾട്ടായി ഉപേക്ഷിക്കാം. ബട്ടണിൽ ക്ലിക്ക് ചെയ്യുക ഡാറ്റാബേസ് സൃഷ്ടിക്കുക:

ത്രെഡ് ഹാൻഡ്ലർ

ഓരോ രണ്ട് സെക്കൻഡിലും കാഫ്കയിൽ നിന്ന് വരുന്ന പുതിയ ഡാറ്റ പ്രോസസ്സ് ചെയ്യുകയും ഡാറ്റാബേസിലേക്ക് ഫലം നൽകുകയും ചെയ്യുന്ന ഒരു സ്പാർക്ക് ജോലിയുടെ വികസനമാണ് അവസാന ഘട്ടം.

മുകളിൽ സൂചിപ്പിച്ചതുപോലെ, തെറ്റ് സഹിഷ്ണുത ഉറപ്പാക്കാൻ കോൺഫിഗർ ചെയ്യേണ്ട സ്പാർക്ക് സ്ട്രീമിംഗിലെ ഒരു പ്രധാന സംവിധാനമാണ് ചെക്ക് പോയിന്റുകൾ. ഞങ്ങൾ ചെക്ക്‌പോസ്റ്റുകൾ ഉപയോഗിക്കും, നടപടിക്രമം പരാജയപ്പെടുകയാണെങ്കിൽ, സ്പാർക്ക് സ്ട്രീമിംഗ് മൊഡ്യൂളിന് അവസാന ചെക്ക് പോയിന്റിലേക്ക് മടങ്ങുകയും അതിൽ നിന്ന് കണക്കുകൂട്ടലുകൾ പുനരാരംഭിക്കുകയും ചെയ്‌താൽ മാത്രം മതി, നഷ്ടപ്പെട്ട ഡാറ്റ വീണ്ടെടുക്കാൻ.

ചെക്ക്‌പോയിന്റ് വിവരങ്ങൾ സംഭരിക്കുന്ന തകരാർ-സഹിഷ്ണുതയുള്ളതും വിശ്വസനീയവുമായ ഫയൽ സിസ്റ്റത്തിൽ (HDFS, S3, മുതലായവ) ഒരു ഡയറക്ടറി സജ്ജീകരിച്ചുകൊണ്ട് ചെക്ക്‌പോയിന്റിംഗ് പ്രവർത്തനക്ഷമമാക്കാം. ഇത് ഉപയോഗിച്ചാണ് ചെയ്യുന്നത്, ഉദാഹരണത്തിന്:

streamingContext.checkpoint(checkpointDirectory)

ഞങ്ങളുടെ ഉദാഹരണത്തിൽ, ഞങ്ങൾ ഇനിപ്പറയുന്ന സമീപനം ഉപയോഗിക്കും, അതായത്, ചെക്ക്‌പോയിന്റ് ഡയറക്‌ടറി നിലവിലുണ്ടെങ്കിൽ, ചെക്ക്‌പോയിന്റ് ഡാറ്റയിൽ നിന്ന് സന്ദർഭം പുനർനിർമ്മിക്കും. ഡയറക്‌ടറി നിലവിലില്ലെങ്കിൽ (അതായത് ആദ്യമായി എക്‌സിക്യൂട്ട് ചെയ്‌തത്), ഒരു പുതിയ സന്ദർഭം സൃഷ്‌ടിക്കുന്നതിനും ഡിസ്‌ട്രീമുകൾ കോൺഫിഗർ ചെയ്യുന്നതിനുമായി ഫംഗ്‌ഷൻToCreateContext-നെ വിളിക്കുന്നു:

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

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 വിന്യസിക്കുന്നു" എന്ന ഘട്ടത്തിൽ ഞങ്ങൾ അതിനായി ഉപയോക്താവും പാസ്‌വേഡും സൃഷ്ടിച്ചു. കണക്റ്റിവിറ്റി & സെക്യൂരിറ്റി വിഭാഗത്തിൽ പ്രദർശിപ്പിക്കുന്ന ഡാറ്റാബേസ് സെർവർ url ആയി നിങ്ങൾ Endpoint ഉപയോഗിക്കണം:

സ്പാർക്കിനെയും കാഫ്കയെയും ശരിയായി ബന്ധിപ്പിക്കുന്നതിന്, ആർട്ടിഫാക്റ്റ് ഉപയോഗിച്ച് സ്മാർക്ക്-സബ്മിറ്റ് വഴി നിങ്ങൾ ജോലി പ്രവർത്തിപ്പിക്കണം. 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 സെക്കൻഡായി സജ്ജമാക്കി:

അടുത്തതായി, പട്ടികയിലെ റെക്കോർഡുകളുടെ സാന്നിധ്യം പരിശോധിക്കാൻ ഞങ്ങൾ ഡാറ്റാബേസിലേക്ക് ഒരു ലളിതമായ അന്വേഷണം നടത്തുന്നു ഇടപാട്_ഫ്ലോ:

തീരുമാനം

Apache Kafka, PostgreSQL എന്നിവയുമായി ചേർന്ന് സ്പാർക്ക് സ്ട്രീമിംഗ് ഉപയോഗിച്ച് വിവരങ്ങളുടെ സ്ട്രീം പ്രോസസ്സിംഗിന്റെ ഒരു ഉദാഹരണം ഈ ലേഖനം പരിശോധിച്ചു. വിവിധ സ്രോതസ്സുകളിൽ നിന്നുള്ള ഡാറ്റയുടെ വളർച്ചയോടെ, സ്ട്രീമിംഗും തത്സമയ ആപ്ലിക്കേഷനുകളും സൃഷ്ടിക്കുന്നതിനുള്ള സ്പാർക്ക് സ്ട്രീമിംഗിന്റെ പ്രായോഗിക മൂല്യം അമിതമായി കണക്കാക്കുന്നത് ബുദ്ധിമുട്ടാണ്.

നിങ്ങൾക്ക് എന്റെ ശേഖരത്തിൽ മുഴുവൻ സോഴ്‌സ് കോഡും കണ്ടെത്താനാകും സാമൂഹികം.

ഈ ലേഖനം ചർച്ച ചെയ്യുന്നതിൽ എനിക്ക് സന്തോഷമുണ്ട്, നിങ്ങളുടെ അഭിപ്രായങ്ങൾക്കായി ഞാൻ പ്രതീക്ഷിക്കുന്നു, കൂടാതെ കരുതലുള്ള എല്ലാ വായനക്കാരിൽ നിന്നും ക്രിയാത്മകമായ വിമർശനവും ഞാൻ പ്രതീക്ഷിക്കുന്നു.

ഞാൻ നിന്നെ വിജയം ആഗ്രഹിക്കുന്നു!

സങ്കീ. തുടക്കത്തിൽ ഒരു പ്രാദേശിക PostgreSQL ഡാറ്റാബേസ് ഉപയോഗിക്കാൻ പദ്ധതിയിട്ടിരുന്നു, എന്നാൽ AWS-നോടുള്ള എന്റെ ഇഷ്ടം കണക്കിലെടുത്ത്, ഡാറ്റാബേസ് ക്ലൗഡിലേക്ക് മാറ്റാൻ ഞാൻ തീരുമാനിച്ചു. ഈ വിഷയത്തെക്കുറിച്ചുള്ള അടുത്ത ലേഖനത്തിൽ, AWS Kinesis, AWS EMR എന്നിവ ഉപയോഗിച്ച് AWS-ൽ മുകളിൽ വിവരിച്ച മുഴുവൻ സിസ്റ്റവും എങ്ങനെ നടപ്പിലാക്കാമെന്ന് ഞാൻ കാണിക്കും. വാർത്ത പിന്തുടരുക!

അവലംബം: www.habr.com