Apache Kafka နှင့် Spark Streaming ဖြင့် လွှင့်ထုတ်သည့် ဒေတာကို လုပ်ဆောင်ခြင်း။

မင်္ဂလာပါ Habr။ ယနေ့ကျွန်ုပ်တို့သည် Spark Streaming ကိုအသုံးပြု၍ Apache Kafka မက်ဆေ့ခ်ျစီးကြောင်းများကိုလုပ်ဆောင်မည့်စနစ်တစ်ခုတည်ဆောက်ပြီး AWS RDS cloud ဒေတာဘေ့စ်တွင်လုပ်ဆောင်ခြင်းရလဒ်များကိုရေးသားပါမည်။

အချို့သော ခရက်ဒစ်အဖွဲ့အစည်းတစ်ခုသည် ၎င်း၏အကိုင်းအခက်များအားလုံးကို “အလျင်အမြန်” လုပ်ဆောင်ရန် ကျွန်ုပ်တို့အား ပေးဆောင်ရမည့်တာဝန်ကို စိတ်ကူးကြည့်ကြပါစို့။ ဘဏ္ဍာတိုက်၊ ကန့်သတ်ချက်များ သို့မဟုတ် ငွေပေးငွေယူများအတွက် ငွေကြေးရလဒ်များ စသည်တို့အတွက် ပွင့်လင်းသော ငွေကြေးအနေအထားကို ချက်ခြင်းတွက်ချက်ရန် ရည်ရွယ်ချက်အတွက် ၎င်းကို လုပ်ဆောင်နိုင်သည်။

မှော်ပညာနှင့် မှော်စာလုံးပေါင်းများအသုံးမပြုဘဲ ဤအမှုကို မည်သို့အကောင်အထည်ဖော်ရမည်နည်း - ဖြတ်တောက်မှုအောက်တွင်ဖတ်ပါ။ သွား!

(ပုံအရင်းအမြစ်)

နိဒါန်း

မှန်ပါသည်၊ အချိန်နှင့်တပြေးညီ ဒေတာအများအပြားကို စီမံဆောင်ရွက်ခြင်းသည် ခေတ်မီစနစ်များတွင် အသုံးပြုရန် အခွင့်အလမ်းများစွာ ပေးပါသည်။ ဤအတွက် ရေပန်းအစားဆုံး ပေါင်းစပ်မှုများထဲမှ တစ်ခုသည် Apache Kafka နှင့် Spark Streaming ၏ tandem ဖြစ်ပြီး၊ Kafka သည် အဝင်မက်ဆေ့ချ် ပက်ကေ့ခ်ျများကို ဖန်တီးပေးကာ Spark Streaming သည် သတ်မှတ်ထားသော အချိန်ကြားကာလတွင် အဆိုပါ ပက်ကတ်များကို လုပ်ဆောင်သည်။

အပလီကေးရှင်း၏ အမှားခံနိုင်ရည်ကို တိုးမြှင့်ရန်အတွက် စစ်ဆေးရေးဂိတ်များကို အသုံးပြုပါမည်။ ဤယန္တရားဖြင့် Spark Streaming အင်ဂျင်သည် ပျောက်ဆုံးသွားသောဒေတာကို ပြန်လည်ရယူရန် လိုအပ်သောအခါ၊ ၎င်းသည် နောက်ဆုံးစစ်ဆေးရေးဂိတ်သို့ ပြန်သွားပြီး ထိုနေရာမှ တွက်ချက်မှုများကို ပြန်လည်လုပ်ဆောင်ရန်သာ လိုအပ်သည်။

ဗိသုကာစနစ် တီထွင်ခဲ့သည်။

အသုံးပြုထားသော အစိတ်အပိုင်းများ

• Apache Kafka ဖြန့်ဝေထားသော ထုတ်ဝေ-စာရင်းသွင်း စာတိုပေးပို့မှုစနစ်တစ်ခုဖြစ်သည်။ အော့ဖ်လိုင်းနှင့် အွန်လိုင်းစာတို သုံးစွဲမှု နှစ်မျိုးလုံးအတွက် သင့်လျော်သည်။ ဒေတာဆုံးရှုံးမှုကို ကာကွယ်ရန်၊ Kafka မက်ဆေ့ချ်များကို ဒစ်ခ်ပေါ်တွင် သိမ်းဆည်းပြီး အစုအဝေးအတွင်း ထပ်တူပြုပါသည်။ Kafka စနစ်သည် ZooKeeper ထပ်တူပြုခြင်းဝန်ဆောင်မှု၏ထိပ်တွင်တည်ဆောက်ထားသည်။
• Apache Spark လွှင့်ခြင်း။ - တိုက်ရိုက်ထုတ်လွှင့်မှုဒေတာကို လုပ်ဆောင်ရန်အတွက် Spark အစိတ်အပိုင်း။ Spark Streaming module ကို ဒေတာစီးကြောင်းအသေးစား ဒေတာပက်ကေ့ခ်ျများ၏ စဉ်ဆက်မပြတ် အစီအစဥ်အဖြစ် အဓိပ္ပာယ်ဖွင့်ဆိုထားသည့် micro-batch ဗိသုကာကို အသုံးပြု၍ တည်ဆောက်ထားသည်။ Spark Streaming သည် မတူညီသော အရင်းအမြစ်များမှ ဒေတာများကို ယူဆောင်ပြီး ၎င်းကို သေးငယ်သော ပက်ကေ့ခ်ျများအဖြစ် ပေါင်းစပ်ထားသည်။ ပက်ကေ့ဂျ်အသစ်များကို ပုံမှန်အချိန်များတွင် ဖန်တီးပါသည်။ အချိန်ကာလတစ်ခုစီ၏အစတွင်၊ ပက်ကေ့ခ်ျအသစ်တစ်ခုဖန်တီးပြီး ထိုကြားကာလအတွင်းရရှိသောဒေတာအားလုံးကို ပက်ကတ်တွင်ထည့်သွင်းထားသည်။ ကြားကာလ၏အဆုံးတွင်၊ ပက်ကတ်ကြီးထွားမှုရပ်တန့်သွားသည်။ ကြားကာလ၏အရွယ်အစားကို batch interval ဟုခေါ်သော အတိုင်းအတာတစ်ခုဖြင့် ဆုံးဖြတ်သည်။
• Apache Spark SQL - Spark functional programming နှင့် ဆက်စပ်ဆောင်ရွက်မှုကို ပေါင်းစပ်ထားသည်။ Structured data ဆိုသည်မှာ မှတ်တမ်းအားလုံးအတွက် ကွက်လပ်တစ်ခုတည်းပါသော ဒေတာကို ဆိုလိုသည်။ Spark SQL သည် အမျိုးမျိုးသောဖွဲ့စည်းပုံဒေတာရင်းမြစ်များမှ ထည့်သွင်းမှုကို ပံ့ပိုးပေးပြီး schema အချက်အလက်ရရှိမှုကြောင့်၊ လိုအပ်သော မှတ်တမ်းများ၏ အကွက်များကိုသာ ထိရောက်စွာ ပြန်လည်ရယူနိုင်ပြီး DataFrame APIs များကိုလည်း ပံ့ပိုးပေးပါသည်။
• AWS RDS စျေးသက်သာသော cloud-based ဆက်စပ်ဒေတာဘေ့စ်၊ တပ်ဆင်မှု၊ လည်ပတ်မှုနှင့် အတိုင်းအတာကို ရိုးရှင်းလွယ်ကူစေသည့် ဝဘ်ဝန်ဆောင်မှုဖြစ်ပြီး 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 ဖြန့်ဝေမှုနှင့်အတူ ပါလာသော အရန် script ကို အသုံးပြုပါမည်-

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

ZooKeeper အောင်မြင်စွာစတင်ပြီးနောက်၊ သီးခြား terminal တွင် Kafka ဆာဗာကိုဖွင့်ပါ-

bin/kafka-server-start.sh config/server.properties

Transaction ဟုခေါ်သော အကြောင်းအရာအသစ်တစ်ခုကို ဖန်တီးကြပါစို့။

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic transaction

လိုအပ်သော partitions နှင့် replication အရေအတွက်များပါရှိသော ခေါင်းစဉ်တစ်ခုကို ဖန်တီးထားကြောင်း သေချာပါစေ။

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

script ကို run သောအခါ၊ terminal တွင်အောက်ပါစာတိုများကိုကျွန်ုပ်တို့လက်ခံရရှိသည်-

ဆိုလိုသည်မှာ အရာအားလုံးသည် ကျွန်ုပ်တို့အလိုရှိသည့်အတိုင်း အလုပ်လုပ်သည် - ထုတ်လုပ်သူက ကျွန်ုပ်တို့လိုအပ်သည့်အကြောင်းအရာဆီသို့ မက်ဆေ့ခ်ျများထုတ်ပေးပြီး ပေးပို့ပါသည်။
နောက်တစ်ဆင့်မှာ Spark ကို ထည့်သွင်းပြီး ဤမက်ဆေ့ချ်စီးကြောင်းကို လုပ်ဆောင်ရန်ဖြစ်သည်။

Apache Spark ကို ထည့်သွင်းခြင်း။

Apache ကို Spark universal နှင့် high-performance cluster computing platform တစ်ခုဖြစ်သည်။

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 ကို ပြောင်းလဲပြီးနောက် အောက်ပါ command ကို run ပါ။

source ~/.bashrc

AWS PostgreSQL ကို အသုံးပြုခြင်း။

ကျန်ရှိနေသေးသည်မှာ ကျွန်ုပ်တို့သည် ထုတ်လွှင့်မှုများမှ လုပ်ဆောင်ပြီးသား အချက်အလက်များကို အပ်လုဒ်လုပ်မည့် ဒေတာဘေ့စ်ကို ဖြန့်ကျက်ရန်ဖြစ်သည်။ ဤအတွက် ကျွန်ုပ်တို့သည် AWS RDS ဝန်ဆောင်မှုကို အသုံးပြုပါမည်။

AWS ကွန်ဆိုးလ် -> AWS RDS -> ဒေတာဘေ့စ် -> ဒေတာဘေ့စ်ကို ဖန်တီးပါ-

PostgreSQL ကိုရွေးပြီး Next ကိုနှိပ်ပါ။

ဘာဖြစ်လို့လဲဆိုတော့ ဤဥပမာသည် ပညာရေးဆိုင်ရာ ရည်ရွယ်ချက်အတွက်သာဖြစ်သည်၊ ကျွန်ုပ်တို့သည် အခမဲ့ဆာဗာ “အနည်းဆုံး” (အခမဲ့အဆင့်) ကို အသုံးပြုပါမည်။

ထို့နောက်၊ ကျွန်ုပ်တို့သည် Free Tier ဘလောက်တွင် အမှန်ခြစ်တစ်ခုထည့်ထားပြီး၊ ထို့နောက်တွင် ကျွန်ုပ်တို့သည် t2.micro အတန်းအစား အားနည်းသော်လည်း ၎င်းသည် အခမဲ့ဖြစ်ပြီး ကျွန်ုပ်တို့၏လုပ်ငန်းအတွက် အလွန်သင့်လျော်ပါသည်။

နောက်တစ်ခုက အလွန်အရေးကြီးတဲ့ အရာတွေ ဖြစ်တဲ့ ဒေတာဘေ့စ် ဥပမာရဲ့ အမည်၊ မာစတာ အသုံးပြုသူရဲ့ အမည်နဲ့ သူ့ရဲ့ စကားဝှက်။ ဥပမာ- myHabrTest၊ မာစတာအသုံးပြုသူ- habrစကားဝှက်- habr12345 ထို့နောက် Next ခလုတ်ကို နှိပ်ပါ။

နောက်စာမျက်နှာတွင် ကျွန်ုပ်တို့၏ဒေတာဘေ့စ်ဆာဗာ၏ ပြင်ပမှဝင်ရောက်နိုင်မှု (အများပြည်သူသုံးနိုင်မှု) နှင့် ဆိပ်ကမ်းရရှိနိုင်မှုတို့အတွက် တာဝန်ရှိသောဘောင်များ ရှိပါသည်။

ဆိပ်ကမ်း 5432 (PostgreSQL) မှတစ်ဆင့် ကျွန်ုပ်တို့၏ဒေတာဘေ့စ်ဆာဗာသို့ ပြင်ပသို့ဝင်ရောက်ခွင့်ပေးမည့် VPC လုံခြုံရေးအဖွဲ့အတွက် ဆက်တင်အသစ်တစ်ခုကို ဖန်တီးကြပါစို့။
VPC Dashboard -> Security Groups -> လုံခြုံရေးအုပ်စုများ ဖန်တီးခြင်းကဏ္ဍသို့ သီးခြားဘရောက်ဆာဝင်းဒိုးမှ AWS ကွန်ဆိုးလ်သို့ သွားကြပါစို့။

လုံခြုံရေးအဖွဲ့အတွက် အမည်ကို ကျွန်ုပ်တို့သတ်မှတ်ထားသည် - PostgreSQL၊ ဖော်ပြချက်တစ်ခု၊ မည်သည့် VPC သည် ဤအဖွဲ့နှင့် ဆက်စပ်သင့်သည်ကို ညွှန်ပြပြီး ဖန်တီးရန် ခလုတ်ကို နှိပ်ပါ-

အောက်ပါပုံတွင်ပြထားသည့်အတိုင်း အသစ်ဖန်တီးထားသောအုပ်စုအတွက် port 5432 အတွက် Inbound စည်းမျဉ်းများကို ဖြည့်စွက်ပါ။ ဆိပ်ကမ်းကို သင်ကိုယ်တိုင် မသတ်မှတ်နိုင်သော်လည်း Type drop-down list မှ PostgreSQL ကို ရွေးပါ။

အတိအကျပြောရလျှင် တန်ဖိုး ::/0 ဆိုသည်မှာ ကမ္ဘာတစ်ဝှမ်းမှ ဆာဗာသို့ အဝင်အထွက်လမ်းကြောင်းကို ရရှိနိုင်မှုဟု ဆိုလိုသည်၊ ၎င်းသည် အဓိပ္ပါယ်အားဖြင့် လုံးဝမမှန်ပါ၊ သို့သော် နမူနာကို ခွဲခြမ်းစိတ်ဖြာရန်၊ ဤချဉ်းကပ်နည်းကို ကျွန်ုပ်တို့ အသုံးပြုခွင့်ပြုကြပါစို့။

ကျွန်ုပ်တို့တွင် "အဆင့်မြင့်ဆက်တင်များကိုပြင်ဆင်ပါ" ကိုဖွင့်ပြီး VPC လုံခြုံရေးအဖွဲ့များကဏ္ဍတွင် ရွေးချယ်ထားသော ဘရောက်ဆာစာမျက်နှာသို့ ပြန်သွားရန် -> ရှိပြီးသား VPC လုံခြုံရေးအဖွဲ့များကို ရွေးချယ်ပါ -> PostgreSQL-

ထို့နောက် Database options -> Database name -> အမည်ကို သတ်မှတ်ပါ - habrDB.

အရန်သိမ်းခြင်း (အရန်သိမ်းထားချိန် - 0 ရက်များ)၊ စောင့်ကြည့်ခြင်းနှင့် စွမ်းဆောင်ရည်ဆိုင်ရာ ထိုးထွင်းသိမြင်မှုများကို ပုံမှန်အားဖြင့် ပိတ်ခြင်းမှလွဲ၍ ကျန်ဘောင်များကို ကျွန်ုပ်တို့ ချန်ထားနိုင်ပါသည်။ ခလုတ်ကိုနှိပ်ပါ။ ဒေတာဘေ့စ်ဖန်တီးပါ။:

ကြိုးကိုင်သူ

နောက်ဆုံးအဆင့်မှာ Kafka မှလာသော ဒေတာအသစ်များကို နှစ်စက္ကန့်တိုင်း လုပ်ဆောင်ပြီး ရလဒ်ကို ဒေတာဘေ့စ်ထဲသို့ ထည့်သွင်းပေးမည့် Spark အလုပ်၏ နောက်ဆုံးအဆင့် ဖွံ့ဖြိုးတိုးတက်လာမည်ဖြစ်သည်။

အထက်တွင်ဖော်ပြခဲ့သည့်အတိုင်း၊ စစ်ဆေးရေးဂိတ်များသည် အမှားအယွင်းခံနိုင်ရည်ရှိစေရန်အတွက် ပြင်ဆင်သတ်မှတ်ရမည့် SparkStreaming တွင် အဓိကယန္တရားတစ်ခုဖြစ်သည်။ ကျွန်ုပ်တို့သည် စစ်ဆေးရေးဂိတ်များကို အသုံးပြုမည်ဖြစ်ပြီး၊ လုပ်ထုံးလုပ်နည်းမအောင်မြင်ပါက Spark Streaming module သည် နောက်ဆုံးစစ်ဆေးရေးဂိတ်သို့ ပြန်သွားရန်သာ လိုအပ်မည်ဖြစ်ပြီး ပျောက်ဆုံးသွားသောဒေတာကို ပြန်လည်ရယူရန်အတွက် ၎င်းမှ တွက်ချက်မှုများကို ပြန်လည်လုပ်ဆောင်ရန် လိုအပ်မည်ဖြစ်သည်။

စစ်ဆေးရေးဂိတ်အချက်အလက်ကို သိမ်းဆည်းထားမည့် အမှား-ခံနိုင်ရည်ရှိသော၊ ယုံကြည်စိတ်ချရသော ဖိုင်စနစ် (HDFS၊ S3 စသည်ဖြင့်) တွင် လမ်းကြောင်းတစ်ခု သတ်မှတ်ခြင်းဖြင့် စစ်ဆေးခြင်းကို ဖွင့်နိုင်သည်။ ဥပမာအားဖြင့်၊

streamingContext.checkpoint(checkpointDirectory)

ကျွန်ုပ်တို့၏ဥပမာတွင်၊ ကျွန်ုပ်တို့သည် အောက်ပါချဉ်းကပ်နည်းကို အသုံးပြုမည်ဖြစ်ပြီး၊ အတိအကျပြောရလျှင် checkpointDirectory ရှိပါက၊ ထို့နောက် အကြောင်းအရာကို စစ်ဆေးရေးဂိတ်ဒေတာမှ ပြန်လည်ဖန်တီးမည်ဖြစ်သည်။ လမ်းညွှန်ချက်မရှိပါက (ဆိုလိုသည်မှာ ပထမအကြိမ်လုပ်ဆောင်ခဲ့သည်)၊ ထို့နောက် ဆက်စပ်အကြောင်းအရာအသစ်တစ်ခုဖန်တီးရန်နှင့် DStreams ကို configure လုပ်ရန် 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")

Spark SQL ကိုအသုံးပြု၍ ကျွန်ုပ်တို့သည် ရိုးရှင်းသောအုပ်စုဖွဲ့ကာ console တွင်ရလဒ်ကိုပြသသည်-

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

Query စာသားကို ရယူပြီး 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 ကို Deploying” အဆင့်တွင် ၎င်းအတွက် အသုံးပြုသူနှင့် စကားဝှက်ကို ဖန်တီးခဲ့သည်။ ချိတ်ဆက်မှုနှင့် လုံခြုံရေးကဏ္ဍတွင် ပြသထားသည့် ဒေတာဘေ့စ်ဆာဗာ url အဖြစ် Endpoint ကို သင်အသုံးပြုသင့်သည်-

Spark နှင့် Kafka ကို မှန်ကန်စွာ ချိတ်ဆက်နိုင်ရန်၊ artifact ကို အသုံးပြု၍ smark-submit မှတစ်ဆင့် အလုပ်ကို လုပ်ဆောင်သင့်သည် spark-streaming-kafka-0-8_2.11. ထို့အပြင်၊ ကျွန်ုပ်တို့သည် PostgreSQL ဒေတာဘေ့စ်နှင့် အပြန်အလှန်တုံ့ပြန်ရန်အတွက် ရှေးဟောင်းပစ္စည်းတစ်ခုကိုလည်း အသုံးပြုမည်ဖြစ်သည်၊ ၎င်းတို့ကို --packages မှတစ်ဆင့် လွှဲပြောင်းပေးမည်ဖြစ်သည်။

Script ၏ ပြောင်းလွယ်ပြင်လွယ်မှုအတွက်၊ ကျွန်ုပ်တို့သည် မက်ဆေ့ချ်ဆာဗာ၏ အမည်နှင့် ဒေတာလက်ခံရယူလိုသည့် အကြောင်းအရာကို ထည့်သွင်းသည့် ကန့်သတ်ချက်များအဖြစ် ပါ၀င်မည်ဖြစ်သည်။

ထို့ကြောင့်၊ စနစ်၏လုပ်ဆောင်နိုင်စွမ်းကို စတင်စစ်ဆေးရန် အချိန်ရောက်ပြီ-

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

အားလုံးအဆင်ပြေသွားပါပြီ။ အောက်ဖော်ပြပါပုံတွင် သင်မြင်ရသည့်အတိုင်း၊ အပလီကေးရှင်းကိုလည်ပတ်နေချိန်တွင်၊ ကျွန်ုပ်တို့သည် StreamingContext အရာဝတ္ထုကိုဖန်တီးသောအခါတွင် ကျွန်ုပ်တို့သည် အတွဲလိုက်ကြားကာလကို 2 စက္ကန့်အဖြစ်သတ်မှတ်ထားသောကြောင့် ပေါင်းစည်းမှုရလဒ်အသစ်များ 2 စက္ကန့်တိုင်းထွက်ရှိလာပါသည်။

ထို့နောက်၊ ကျွန်ုပ်တို့သည် ဇယားရှိ မှတ်တမ်းများ ရှိနေခြင်းကို စစ်ဆေးရန် ဒေတာဘေ့စ်သို့ ရိုးရှင်းသော စုံစမ်းမှုတစ်ခု ပြုလုပ်သည်။ ငွေပေးငွေယူ_စီးဆင်းမှု:

ကောက်ချက်

ဤဆောင်းပါးသည် Apache Kafka နှင့် PostgreSQL တို့နှင့် တွဲဖက် Spark Streaming ကို အသုံးပြု၍ သတင်းအချက်အလက် တိုက်ရိုက်ထုတ်လွှင့်ခြင်း၏ နမူနာကို လေ့လာခဲ့သည်။ ရင်းမြစ်အမျိုးမျိုးမှ ဒေတာများ တိုးပွားလာသည်နှင့်အမျှ streaming နှင့် real-time applications များဖန်တီးရန်အတွက် Spark Streaming ၏ လက်တွေ့ကျသောတန်ဖိုးကို ကျော်လွန်ခန့်မှန်းရန် ခက်ခဲသည်။

ကျွန်ုပ်၏ repository တွင် အရင်းအမြစ်ကုဒ်အပြည့်အစုံကို သင်ရှာဖွေနိုင်ပါသည်။ GitHub.

ဒီဆောင်းပါးကို ဆွေးနွေးရတာ ဝမ်းသာပါတယ်၊ မင်းရဲ့ မှတ်ချက်တွေကို စောင့်မျှော်နေပြီး ဂရုစိုက်တတ်တဲ့ စာဖတ်သူအားလုံးရဲ့ အပြုသဘောဆောင်တဲ့ ဝေဖန်မှုတွေကို မျှော်လင့်ပါတယ်။

အောင်မြင်ပါစေလို့ ဆုတောင်းပါတယ်။

ဆာ အစပိုင်းမှာတော့ ဒေသတွင်း PostgreSQL ဒေတာဘေ့စ်ကို အသုံးပြုဖို့ စီစဉ်ခဲ့ပေမယ့် AWS ကို ချစ်မြတ်နိုးတာကြောင့် ဒေတာဘေ့စ်ကို cloud သို့ ရွှေ့ဖို့ ဆုံးဖြတ်ခဲ့ပါတယ်။ ဤအကြောင်းအရာအတွက် နောက်ဆောင်းပါးတွင်၊ AWS Kinesis နှင့် AWS EMR ကိုအသုံးပြု၍ အထက်တွင်ဖော်ပြထားသော AWS တွင်ဖော်ပြထားသော စနစ်တစ်ခုလုံးကို မည်သို့အကောင်အထည်ဖော်ရမည်ကို ကျွန်ုပ်ပြသပါမည်။ သတင်းကို လိုက်နာပါ။

source: www.habr.com