Apache Kafka និងដំណើរការទិន្នន័យស្ទ្រីមជាមួយ Spark Streaming

សួស្តី ហាប! ថ្ងៃនេះយើងនឹងបង្កើតប្រព័ន្ធមួយដែលនឹងដំណើរការស្ទ្រីមសារ Apache Kafka ដោយប្រើ Spark Streaming ហើយសរសេរលទ្ធផលដំណើរការទៅមូលដ្ឋានទិន្នន័យពពក AWS RDS ។

ចូរយើងស្រមៃថាស្ថាប័នឥណទានជាក់លាក់មួយកំណត់ឱ្យយើងនូវភារកិច្ចនៃដំណើរការប្រតិបត្តិការចូល "ភ្លាមៗ" នៅទូទាំងសាខាទាំងអស់របស់វា។ នេះអាចត្រូវបានធ្វើសម្រាប់គោលបំណងនៃការគណនាភ្លាមៗនូវទីតាំងរូបិយប័ណ្ណបើកចំហសម្រាប់រតនាគារ ដែនកំណត់ ឬលទ្ធផលហិរញ្ញវត្ថុសម្រាប់ប្រតិបត្តិការ។ល។

តើធ្វើដូចម្តេចដើម្បីអនុវត្តករណីនេះដោយមិនប្រើវេទមន្តនិងអក្ខរាវិរុទ្ធ - អាននៅក្រោមការកាត់! ទៅ!

(ប្រភពរូបភាព)

សេចក្តីណែនាំ

ជាការពិតណាស់ ការដំណើរការទិន្នន័យមួយចំនួនធំក្នុងពេលវេលាពិតផ្តល់នូវឱកាសច្រើនសម្រាប់ប្រើប្រាស់ក្នុងប្រព័ន្ធទំនើប។ ការរួមបញ្ចូលគ្នាដ៏ពេញនិយមបំផុតមួយសម្រាប់ការនេះគឺការភ្ជាប់គ្នានៃ Apache Kafka និង Spark Streaming ដែល Kafka បង្កើតស្ទ្រីមនៃកញ្ចប់សារចូល ហើយ Spark Streaming ដំណើរការកញ្ចប់ព័ត៌មានទាំងនេះនៅចន្លោះពេលជាក់លាក់មួយ។

ដើម្បីបង្កើនភាពអត់ធ្មត់នៃកំហុសនៃកម្មវិធី យើងនឹងប្រើចំណុចត្រួតពិនិត្យ។ ជាមួយនឹងយន្តការនេះ នៅពេលដែលម៉ាស៊ីន Spark Streaming ត្រូវការសង្គ្រោះទិន្នន័យដែលបាត់នោះ វាគ្រាន់តែត្រូវត្រលប់ទៅកន្លែងត្រួតពិនិត្យចុងក្រោយ ហើយបន្តការគណនាឡើងវិញពីទីនោះ។

ស្ថាបត្យកម្មនៃប្រព័ន្ធអភិវឌ្ឍន៍

សមាសធាតុដែលបានប្រើ៖

• Apache Kafka គឺ​ជា​ប្រព័ន្ធ​ផ្ញើ​សារ​ដែល​បាន​ចែក​ចាយ​ផ្សាយ​ជាវ។ ស័ក្តិសមសម្រាប់ការប្រើប្រាស់សារក្រៅបណ្តាញ និងអ៊ីនធឺណិត។ ដើម្បីទប់ស្កាត់ការបាត់បង់ទិន្នន័យ សារ Kafka ត្រូវបានរក្សាទុកនៅលើថាស និងចម្លងនៅក្នុងចង្កោម។ ប្រព័ន្ធ Kafka ត្រូវបានបង្កើតឡើងនៅលើកំពូលនៃសេវាកម្មធ្វើសមកាលកម្ម ZooKeeper ។
• ស្ទ្រីម Apache Spark - សមាសធាតុ Spark សម្រាប់ដំណើរការទិន្នន័យស្ទ្រីម។ ម៉ូឌុល Spark Streaming ត្រូវបានបង្កើតឡើងដោយប្រើប្រាស់ស្ថាបត្យកម្មខ្នាតតូចដែលស្ទ្រីមទិន្នន័យត្រូវបានបកស្រាយថាជាបណ្តុំបន្តនៃកញ្ចប់ទិន្នន័យតូចៗ។ Spark Streaming យកទិន្នន័យពីប្រភពផ្សេងៗគ្នា ហើយបញ្ចូលវាទៅក្នុងកញ្ចប់តូចៗ។ កញ្ចប់ថ្មីត្រូវបានបង្កើតនៅចន្លោះពេលទៀងទាត់។ នៅដើមនៃចន្លោះពេលនីមួយៗ កញ្ចប់ព័ត៌មានថ្មីត្រូវបានបង្កើត ហើយទិន្នន័យណាមួយដែលទទួលបានក្នុងចន្លោះពេលនោះត្រូវបានបញ្ចូលក្នុងកញ្ចប់ព័ត៌មាន។ នៅចុងបញ្ចប់នៃចន្លោះពេល ការលូតលាស់កញ្ចប់ព័ត៌មានឈប់។ ទំហំនៃចន្លោះពេលត្រូវបានកំណត់ដោយប៉ារ៉ាម៉ែត្រដែលហៅថាចន្លោះពេលបាច់;
• កម្មវិធី Apache Spark SQL - រួមបញ្ចូលគ្នានូវដំណើរការទំនាក់ទំនងជាមួយនឹងការសរសេរកម្មវិធីមុខងារ Spark ។ ទិន្នន័យដែលមានរចនាសម្ព័ន្ធមានន័យថាទិន្នន័យដែលមានគ្រោងការណ៍ ពោលគឺសំណុំវាលតែមួយសម្រាប់កំណត់ត្រាទាំងអស់។ Spark SQL គាំទ្រការបញ្ចូលពីប្រភពទិន្នន័យដែលមានរចនាសម្ព័ន្ធផ្សេងៗគ្នា ហើយដោយសារភាពអាចរកបាននៃព័ត៌មានគ្រោងការណ៍ វាអាចទាញយកបានយ៉ាងមានប្រសិទ្ធភាពត្រឹមតែវាលដែលត្រូវការនៃកំណត់ត្រាប៉ុណ្ណោះ ហើយក៏ផ្តល់នូវ DataFrame APIs ផងដែរ។
• AWS RDS គឺជាមូលដ្ឋានទិន្នន័យទំនាក់ទំនងដែលមានមូលដ្ឋានលើពពកដែលមានតំលៃថោក សេវាកម្មគេហទំព័រដែលសម្រួលការដំឡើង ប្រតិបត្តិការ និងការធ្វើមាត្រដ្ឋាន ហើយត្រូវបានគ្រប់គ្រងដោយផ្ទាល់ដោយ Amazon ។

ការដំឡើងនិងដំណើរការម៉ាស៊ីនមេ Kafka

មុននឹងប្រើ Kafka ដោយផ្ទាល់ អ្នកត្រូវតែប្រាកដថាអ្នកមាន Java ព្រោះ... JVM ត្រូវបានប្រើសម្រាប់ការងារ៖

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

តោះបង្កើតអ្នកប្រើប្រាស់ថ្មីដើម្បីធ្វើការជាមួយ Kafka៖

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

មុនពេលចាប់ផ្តើមម៉ាស៊ីនមេ Kafka អ្នកត្រូវចាប់ផ្តើមម៉ាស៊ីនមេ ZooKeeper យើងនឹងប្រើស្គ្រីបជំនួយដែលភ្ជាប់មកជាមួយការចែកចាយ Kafka៖

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 ហើយចុច Next:

ដោយសារតែ ឧទាហរណ៍នេះគឺសម្រាប់គោលបំណងអប់រំតែប៉ុណ្ណោះ យើងនឹងប្រើម៉ាស៊ីនមេឥតគិតថ្លៃ “យ៉ាងហោចណាស់” (ថ្នាក់ឥតគិតថ្លៃ)៖

បន្ទាប់មក យើងដាក់សញ្ញាធីកនៅក្នុងប្លុក Free Tier ហើយបន្ទាប់ពីនោះ យើងនឹងត្រូវបានផ្តល់ជូនដោយស្វ័យប្រវត្តិនូវឧទាហរណ៍នៃថ្នាក់ t2.micro - ទោះបីជាខ្សោយក៏ដោយ វាគឺឥតគិតថ្លៃ និងសមរម្យសម្រាប់កិច្ចការរបស់យើង៖

បន្ទាប់​មក​រឿង​សំខាន់​ខ្លាំង​ណាស់៖ ឈ្មោះ​នៃ​ឧទាហរណ៍​មូលដ្ឋាន​ទិន្នន័យ ឈ្មោះ​អ្នក​ប្រើ​មេ និង​ពាក្យ​សម្ងាត់​របស់​គាត់។ តោះដាក់ឈ្មោះឧទាហរណ៍៖ myHabrTest អ្នកប្រើប្រាស់មេ៖ ហាប់, ពាក្យសម្ងាត់៖ habr12345 ហើយចុចលើប៊ូតុងបន្ទាប់៖

នៅទំព័របន្ទាប់មានប៉ារ៉ាម៉ែត្រដែលទទួលខុសត្រូវចំពោះភាពងាយស្រួលនៃម៉ាស៊ីនមេមូលដ្ឋានទិន្នន័យរបស់យើងពីខាងក្រៅ (លទ្ធភាពប្រើប្រាស់សាធារណៈ) និងលទ្ធភាពច្រក៖

តោះបង្កើតការកំណត់ថ្មីសម្រាប់ក្រុមសុវត្ថិភាព VPC ដែលនឹងអនុញ្ញាតឱ្យចូលប្រើខាងក្រៅទៅកាន់ម៉ាស៊ីនមេមូលដ្ឋានទិន្នន័យរបស់យើងតាមរយៈច្រក 5432 (PostgreSQL)។
តោះទៅកុងសូល AWS នៅក្នុងបង្អួចកម្មវិធីរុករកដាច់ដោយឡែកទៅកាន់ VPC Dashboard -> ក្រុមសុវត្ថិភាព -> បង្កើតផ្នែកក្រុមសុវត្ថិភាព៖

យើងកំណត់ឈ្មោះសម្រាប់ក្រុមសន្តិសុខ - PostgreSQL ដែលជាការពិពណ៌នាបង្ហាញថា VPC ក្រុមនេះគួរត្រូវបានភ្ជាប់ជាមួយ ហើយចុចប៊ូតុងបង្កើត៖

បំពេញច្បាប់ Inbound សម្រាប់ច្រក 5432 សម្រាប់ក្រុមដែលទើបបង្កើតថ្មី ដូចបង្ហាញក្នុងរូបភាពខាងក្រោម។ អ្នកមិនអាចបញ្ជាក់ច្រកដោយដៃបានទេ ប៉ុន្តែជ្រើសរើស PostgreSQL ពីបញ្ជីទម្លាក់ចុះ ប្រភេទ។

និយាយយ៉ាងតឹងរឹង តម្លៃ ::/0 មានន័យថាលទ្ធភាពនៃចរាចរចូលទៅកាន់ម៉ាស៊ីនមេពីគ្រប់ទិសទី ដែលមិនពិតទាំងស្រុង ប៉ុន្តែដើម្បីវិភាគឧទាហរណ៍ សូមអនុញ្ញាតឱ្យខ្លួនយើងប្រើវិធីសាស្រ្តនេះ៖

យើងត្រឡប់ទៅទំព័រកម្មវិធីរុករកវិញ ដែលយើងបើក "កំណត់រចនាសម្ព័ន្ធកម្រិតខ្ពស់" ហើយជ្រើសរើសនៅក្នុងផ្នែកក្រុមសុវត្ថិភាព VPC -> ជ្រើសរើសក្រុមសុវត្ថិភាព VPC ដែលមានស្រាប់ -> PostgreSQL:

បន្ទាប់នៅក្នុងជម្រើសមូលដ្ឋានទិន្នន័យ -> ឈ្មោះមូលដ្ឋានទិន្នន័យ -> កំណត់ឈ្មោះ - habrDB.

យើងអាចទុកប៉ារ៉ាម៉ែត្រដែលនៅសល់ លើកលែងតែការបិទការបម្រុងទុក (រយៈពេលរក្សាទុកការបម្រុងទុក - 0 ថ្ងៃ) ការត្រួតពិនិត្យ និងការអនុវត្តការយល់ដឹងតាមលំនាំដើម។ ចុចលើប៊ូតុង បង្កើតមូលដ្ឋានទិន្នន័យ:

អ្នកដោះស្រាយខ្សែស្រឡាយ

ដំណាក់កាលចុងក្រោយនឹងជាការអភិវឌ្ឍន៍ការងារ Spark ដែលនឹងដំណើរការទិន្នន័យថ្មីដែលមកពី Kafka រៀងរាល់ពីរវិនាទីម្តង ហើយបញ្ចូលលទ្ធផលទៅក្នុងមូលដ្ឋានទិន្នន័យ។

ដូចដែលបានកត់សម្គាល់ខាងលើ ចំណុចត្រួតពិនិត្យគឺជាយន្តការស្នូលមួយនៅក្នុង SparkStreaming ដែលត្រូវតែកំណត់រចនាសម្ព័ន្ធដើម្បីធានាបាននូវការអត់ឱនចំពោះកំហុស។ យើងនឹងប្រើប៉ុស្តិ៍ត្រួតពិនិត្យ ហើយប្រសិនបើដំណើរការបរាជ័យ ម៉ូឌុល Spark Streaming នឹងគ្រាន់តែត្រូវត្រលប់ទៅចំណុចត្រួតពិនិត្យចុងក្រោយ ហើយបន្តការគណនាឡើងវិញពីវាដើម្បីសង្គ្រោះទិន្នន័យដែលបាត់បង់។

ការត្រួតពិនិត្យអាចត្រូវបានបើកដោយការកំណត់ថតនៅលើប្រព័ន្ធឯកសារដែលអាចទុកចិត្តបាននិងអត់ឱនកំហុស (ដូចជា HDFS, S3 ។ល។) ដែលព័ត៌មានចំណុចត្រួតពិនិត្យនឹងត្រូវបានរក្សាទុក។ នេះត្រូវបានធ្វើដោយប្រើឧទាហរណ៍៖

streamingContext.checkpoint(checkpointDirectory)

ក្នុងឧទាហរណ៍របស់យើង យើងនឹងប្រើវិធីសាស្រ្តខាងក្រោម ពោលគឺប្រសិនបើ checkpointDirectory មាន នោះបរិបទនឹងត្រូវបានបង្កើតឡើងវិញពីទិន្នន័យចំណុចត្រួតពិនិត្យ។ ប្រសិនបើថតមិនមានទេ (ឧ. ប្រតិបត្តិជាលើកដំបូង) បន្ទាប់មក functionToCreateContext ត្រូវបានហៅដើម្បីបង្កើតបរិបទថ្មី និងកំណត់រចនាសម្ព័ន្ធ DStreams៖

from pyspark.streaming import StreamingContext

context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

យើងបង្កើតវត្ថុ DirectStream ដើម្បីភ្ជាប់ទៅប្រធានបទ "ប្រតិបត្តិការ" ដោយប្រើវិធីសាស្ត្រ createDirectStream នៃបណ្ណាល័យ KafkaUtils៖

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 ផងដែរ យើងនឹងផ្ទេរពួកវាតាមរយៈ --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 វិនាទី ពីព្រោះយើងកំណត់ចន្លោះពេលបាច់ទៅ 2 វិនាទីនៅពេលយើងបង្កើតវត្ថុ StreamingContext៖

បន្ទាប់មក យើងធ្វើសំណួរសាមញ្ញមួយទៅកាន់មូលដ្ឋានទិន្នន័យ ដើម្បីពិនិត្យមើលវត្តមាននៃកំណត់ត្រានៅក្នុងតារាង លំហូរប្រតិបត្តិការ:

សេចក្តីសន្និដ្ឋាន

អត្ថបទនេះបានក្រឡេកមើលឧទាហរណ៍នៃដំណើរការស្ទ្រីមព័ត៌មានដោយប្រើ Spark Streaming ដោយភ្ជាប់ជាមួយ Apache Kafka និង PostgreSQL ។ ជាមួយនឹងការកើនឡើងនៃទិន្នន័យពីប្រភពផ្សេងៗ វាពិបាកក្នុងការប៉ាន់ស្មានតម្លៃជាក់ស្តែងនៃ Spark Streaming សម្រាប់ការបង្កើតការផ្សាយ និងកម្មវិធីក្នុងពេលជាក់ស្តែង។

អ្នកអាចស្វែងរកកូដប្រភពពេញលេញនៅក្នុងឃ្លាំងរបស់ខ្ញុំនៅ GitHub.

ខ្ញុំរីករាយក្នុងការពិភាក្សាអត្ថបទនេះ ខ្ញុំទន្ទឹងរង់ចាំយោបល់របស់អ្នក ហើយខ្ញុំក៏សង្ឃឹមសម្រាប់ការរិះគន់ក្នុងន័យស្ថាបនាពីអ្នកអានដែលយកចិត្តទុកដាក់ទាំងអស់។

ខ្ញុំសូមជូនពរឱ្យអ្នកទទួលបានជោគជ័យ!

ទំនុក។ ដំបូងវាត្រូវបានគ្រោងនឹងប្រើមូលដ្ឋានទិន្នន័យ PostgreSQL ក្នុងស្រុក ប៉ុន្តែដោយសារការស្រលាញ់របស់ខ្ញុំចំពោះ AWS ខ្ញុំបានសម្រេចចិត្តផ្លាស់ទីមូលដ្ឋានទិន្នន័យទៅពពក។ នៅក្នុងអត្ថបទបន្ទាប់លើប្រធានបទនេះ ខ្ញុំនឹងបង្ហាញពីរបៀបអនុវត្តប្រព័ន្ធទាំងមូលដែលបានពិពណ៌នាខាងលើនៅក្នុង AWS ដោយប្រើ AWS Kinesis និង AWS EMR ។ តាមដានព័ត៌មាន!

ប្រភព: www.habr.com