Apache Kafka ແລະການປະມວນຜົນຂໍ້ມູນການຖ່າຍທອດດ້ວຍ Spark Streaming

ສະບາຍດີ, Habr! ມື້ນີ້ພວກເຮົາຈະສ້າງລະບົບທີ່ຈະປະມວນຜົນການຖ່າຍທອດຂໍ້ຄວາມ Apache Kafka ໂດຍໃຊ້ Spark Streaming ແລະຂຽນຜົນການປະມວນຜົນໃສ່ຖານຂໍ້ມູນ AWS RDS cloud.

ຂໍໃຫ້ຈິນຕະນາການວ່າສະຖາບັນສິນເຊື່ອສະເພາະໃດຫນຶ່ງກໍານົດໃຫ້ພວກເຮົາປະຕິບັດຫນ້າທຸລະກໍາທີ່ເຂົ້າມາ "ໃນທັນທີ" ໃນທົ່ວສາຂາຂອງມັນ. ນີ້ສາມາດເຮັດໄດ້ເພື່ອຈຸດປະສົງຂອງການຄິດໄລ່ທັນທີທັນໃດຕໍາແຫນ່ງເງິນຕາເປີດສໍາລັບຄັງເງິນ, ຂອບເຂດຈໍາກັດຫຼືຜົນໄດ້ຮັບທາງດ້ານການເງິນສໍາລັບການເຮັດທຸລະກໍາ, ແລະອື່ນໆ.

ວິທີການປະຕິບັດກໍລະນີນີ້ໂດຍບໍ່ມີການໃຊ້ magic ແລະ magic ການສະກົດຄໍາ - ອ່ານພາຍໃຕ້ການຕັດ! ໄປ!

(ແຫຼ່ງຮູບ)

ການນໍາສະເຫນີ

ແນ່ນອນ, ການປະມວນຜົນຂໍ້ມູນຈໍານວນຫຼວງຫຼາຍໃນເວລາທີ່ແທ້ຈິງໃຫ້ໂອກາດອັນພຽງພໍສໍາລັບການນໍາໃຊ້ໃນລະບົບທີ່ທັນສະໄຫມ. ຫນຶ່ງໃນການປະສົມປະສານທີ່ນິຍົມທີ່ສຸດສໍາລັບການນີ້ແມ່ນ tandem ຂອງ Apache Kafka ແລະ Spark Streaming, ທີ່ Kafka ສ້າງກະແສຂອງຊອງຂໍ້ຄວາມທີ່ເຂົ້າມາ, ແລະ Spark Streaming ປະມວນຜົນຊອງເຫຼົ່ານີ້ໃນຊ່ວງເວລາໃດຫນຶ່ງ.

ເພື່ອເພີ່ມຄວາມທົນທານຄວາມຜິດຂອງຄໍາຮ້ອງສະຫມັກ, ພວກເຮົາຈະນໍາໃຊ້ຈຸດກວດກາ. ດ້ວຍກົນໄກນີ້, ເມື່ອເຄື່ອງຈັກ Spark Streaming ຕ້ອງການກູ້ຄືນຂໍ້ມູນທີ່ສູນຫາຍ, ມັນພຽງແຕ່ຕ້ອງການກັບຄືນໄປຫາຈຸດກວດກາສຸດທ້າຍແລະສືບຕໍ່ການຄິດໄລ່ຈາກບ່ອນນັ້ນ.

ສະຖາປັດຕະຍະກໍາຂອງລະບົບການພັດທະນາ

ອົງ​ປະ​ກອບ​ທີ່​ນໍາ​ໃຊ້​:

• Apache Kafka ແມ່ນລະບົບການສົ່ງຂໍ້ຄວາມທີ່ເຜີຍແຜ່-ສະໝັກສະມາຊິກ. ເຫມາະສໍາລັບການບໍລິໂພກຂໍ້ຄວາມອອບໄລນ໌ແລະອອນໄລນ໌. ເພື່ອປ້ອງກັນການສູນເສຍຂໍ້ມູນ, ຂໍ້ຄວາມ Kafka ຖືກເກັບໄວ້ໃນແຜ່ນແລະ replicated ພາຍໃນ cluster. ລະບົບ Kafka ຖືກສ້າງຂຶ້ນຢູ່ເທິງສຸດຂອງການບໍລິການ synchronization ZooKeeper;
• Apache Spark Streaming - ອົງປະກອບ Spark ສໍາລັບການປະມວນຜົນຂໍ້ມູນ streaming. ໂມດູນ Spark Streaming ຖືກສ້າງຂຶ້ນໂດຍໃຊ້ສະຖາປັດຕະຍະກໍາຈຸນລະພາກ, ບ່ອນທີ່ການຖ່າຍທອດຂໍ້ມູນຖືກຕີຄວາມວ່າເປັນລໍາດັບຢ່າງຕໍ່ເນື່ອງຂອງຊຸດຂໍ້ມູນຂະຫນາດນ້ອຍ. Spark Streaming ເອົາຂໍ້ມູນຈາກແຫຼ່ງຕ່າງໆ ແລະລວມມັນເຂົ້າໄປໃນຊຸດນ້ອຍໆ. ແພັກເກດໃຫມ່ຖືກສ້າງຂື້ນໃນຊ່ວງເວລາປົກກະຕິ. ໃນຕອນເລີ່ມຕົ້ນຂອງແຕ່ລະຊ່ວງເວລາ, ແພັກເກັດໃໝ່ຖືກສ້າງຂື້ນ, ແລະຂໍ້ມູນທີ່ໄດ້ຮັບໃນລະຫວ່າງນັ້ນແມ່ນລວມຢູ່ໃນແພັກເກັດ. ໃນຕອນທ້າຍຂອງໄລຍະຫ່າງ, ການຂະຫຍາຍຕົວຂອງແພັກເກັດຢຸດ. ຂະຫນາດຂອງໄລຍະຫ່າງແມ່ນຖືກກໍານົດໂດຍພາລາມິເຕີທີ່ເອີ້ນວ່າ batch interval;
• Apache Spark SQL - ປະ​ສົມ​ປະ​ສານ​ການ​ປະ​ມວນ​ຜົນ​ທີ່​ກ່ຽວ​ຂ້ອງ​ກັບ Spark ການ​ດໍາ​ເນີນ​ໂຄງ​ການ​ທີ່​ເປັນ​ປະ​ໂຫຍດ​. ຂໍ້ມູນທີ່ມີໂຄງສ້າງຫມາຍຄວາມວ່າຂໍ້ມູນທີ່ມີ schema, ນັ້ນແມ່ນ, ຊຸດດຽວຂອງພາກສະຫນາມສໍາລັບບັນທຶກທັງຫມົດ. Spark SQL ສະຫນັບສະຫນູນການປ້ອນຂໍ້ມູນຈາກຫຼາຍໆແຫຼ່ງຂໍ້ມູນທີ່ມີໂຄງສ້າງແລະ, ຍ້ອນການມີຂໍ້ມູນຂອງ schema, ມັນສາມາດດຶງຂໍ້ມູນພຽງແຕ່ພາກສະຫນາມທີ່ກໍານົດໄວ້ຢ່າງມີປະສິດທິພາບ, ແລະຍັງສະຫນອງ 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 ໃນ terminal ແຍກຕ່າງຫາກ:

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

ເມື່ອແລ່ນສະຄຣິບ, ພວກເຮົາໄດ້ຮັບຂໍ້ຄວາມຕໍ່ໄປນີ້ຢູ່ໃນ terminal:

ນີ້ຫມາຍຄວາມວ່າທຸກສິ່ງທຸກຢ່າງເຮັດວຽກຕາມທີ່ພວກເຮົາຕ້ອງການ - ຜູ້ຜະລິດສ້າງແລະສົ່ງຂໍ້ຄວາມໄປຫາຫົວຂໍ້ທີ່ພວກເຮົາຕ້ອງການ.
ຂັ້ນຕອນຕໍ່ໄປແມ່ນການຕິດຕັ້ງ 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):

ຕໍ່ໄປ, ພວກເຮົາເອົາຫມາຍຕິກໃສ່ໃນ Free Tier block, ແລະຫຼັງຈາກນັ້ນພວກເຮົາຈະໄດ້ຮັບການສະເຫນີອັດຕະໂນມັດຂອງ t2.micro class - ເຖິງແມ່ນວ່າອ່ອນແອ, ມັນແມ່ນບໍ່ເສຍຄ່າແລະຂ້ອນຂ້າງເຫມາະສົມສໍາລັບວຽກງານຂອງພວກເຮົາ:

ຕໍ່ໄປສິ່ງທີ່ມີຄວາມສໍາຄັນຫຼາຍ: ຊື່ຂອງຕົວຢ່າງຖານຂໍ້ມູນ, ຊື່ຂອງຜູ້ໃຊ້ຕົ້ນສະບັບແລະລະຫັດຜ່ານຂອງລາວ. ໃຫ້ຕັ້ງຊື່ຕົວຢ່າງ: myHabrTest, ຜູ້ໃຊ້ຕົ້ນສະບັບ: habr, ລະ​ຫັດ​ຜ່ານ​: habr12345 ແລະ​ໃຫ້​ຄລິກ​ໃສ່​ປຸ່ມ​ຕໍ່​ໄປ​:

ໃນຫນ້າຕໍ່ໄປມີພາລາມິເຕີທີ່ຮັບຜິດຊອບສໍາລັບການເຂົ້າເຖິງຂອງເຄື່ອງແມ່ຂ່າຍຖານຂໍ້ມູນຂອງພວກເຮົາຈາກພາຍນອກ (ການເຂົ້າເຖິງສາທາລະນະ) ແລະຄວາມພ້ອມຂອງພອດ:

ໃຫ້ສ້າງການຕັ້ງຄ່າໃຫມ່ສໍາລັບກຸ່ມຄວາມປອດໄພ VPC, ເຊິ່ງຈະຊ່ວຍໃຫ້ການເຂົ້າເຖິງພາຍນອກກັບເຄື່ອງແມ່ຂ່າຍຖານຂໍ້ມູນຂອງພວກເຮົາຜ່ານພອດ 5432 (PostgreSQL).
ໃຫ້ໄປທີ່ຄອນໂຊ AWS ໃນປ່ອງຢ້ຽມຂອງຕົວທ່ອງເວັບແຍກຕ່າງຫາກໄປຫາ VPC Dashboard -> ກຸ່ມຄວາມປອດໄພ -> ສ້າງກຸ່ມຄວາມປອດໄພ:

ພວກເຮົາຕັ້ງຊື່ສໍາລັບກຸ່ມຄວາມປອດໄພ - PostgreSQL, ຄໍາອະທິບາຍ, ຊີ້ບອກວ່າ VPC ກຸ່ມນີ້ຄວນຈະຖືກເຊື່ອມໂຍງກັບແລະກົດປຸ່ມສ້າງ:

ຕື່ມຂໍ້ມູນໃສ່ກົດລະບຽບຂາເຂົ້າສໍາລັບພອດ 5432 ສໍາລັບກຸ່ມທີ່ສ້າງໃຫມ່, ດັ່ງທີ່ສະແດງຢູ່ໃນຮູບຂ້າງລຸ່ມນີ້. ທ່ານບໍ່ສາມາດລະບຸພອດດ້ວຍຕົນເອງ, ແຕ່ເລືອກ PostgreSQL ຈາກລາຍການແບບເລື່ອນລົງປະເພດ.

ເວົ້າຢ່າງເຂັ້ມງວດ, ມູນຄ່າ ::/0 ຫມາຍເຖິງການມີການຈະລາຈອນເຂົ້າມາກັບເຄື່ອງແມ່ຂ່າຍຈາກທົ່ວໂລກ, ເຊິ່ງບໍ່ແມ່ນຄວາມຈິງທັງຫມົດ, ແຕ່ເພື່ອວິເຄາະຕົວຢ່າງ, ໃຫ້ພວກເຮົາອະນຸຍາດໃຫ້ໃຊ້ວິທີການນີ້:

ພວກເຮົາກັບຄືນໄປຫາຫນ້າຕົວທ່ອງເວັບ, ບ່ອນທີ່ພວກເຮົາມີ "ຕັ້ງຄ່າການຕັ້ງຄ່າຂັ້ນສູງ" ເປີດແລະເລືອກໃນພາກກຸ່ມຄວາມປອດໄພ VPC -> ເລືອກກຸ່ມຄວາມປອດໄພ VPC ທີ່ມີຢູ່ແລ້ວ -> PostgreSQL:

ຕໍ່ໄປ, ໃນຕົວເລືອກຖານຂໍ້ມູນ -> ຊື່ຖານຂໍ້ມູນ -> ຕັ້ງຊື່ - habrDB.

ພວກເຮົາສາມາດອອກຈາກຕົວກໍານົດການທີ່ຍັງເຫຼືອ, ຍົກເວັ້ນການປິດການສໍາຮອງຂໍ້ມູນ (ໄລຍະເວລາເກັບຮັກສາສໍາຮອງຂໍ້ມູນ - 0 ມື້), ການຕິດຕາມແລະການປະຕິບັດ Insights, ໂດຍຄ່າເລີ່ມຕົ້ນ. ໃຫ້ຄລິກໃສ່ປຸ່ມ ສ້າງຖານຂໍ້ມູນ:

ຕົວຈັດການກະທູ້

ຂັ້ນຕອນສຸດທ້າຍຈະເປັນການພັດທະນາວຽກ Spark, ເຊິ່ງຈະປະມວນຜົນຂໍ້ມູນໃຫມ່ທີ່ມາຈາກ Kafka ທຸກໆສອງວິນາທີແລະໃສ່ຜົນໄດ້ຮັບເຂົ້າໄປໃນຖານຂໍ້ມູນ.

ດັ່ງທີ່ໄດ້ກ່າວໄວ້ຂ້າງເທິງ, ຈຸດກວດກາແມ່ນກົນໄກຫຼັກໃນ SparkStreaming ທີ່ຕ້ອງໄດ້ຮັບການຕັ້ງຄ່າເພື່ອຮັບປະກັນຄວາມທົນທານຕໍ່ຄວາມຜິດພາດ. ພວກເຮົາຈະນໍາໃຊ້ຈຸດກວດກາແລະ, ຖ້າຂັ້ນຕອນລົ້ມເຫລວ, ໂມດູນ Spark Streaming ພຽງແຕ່ຕ້ອງການກັບຄືນໄປຫາຈຸດກວດກາສຸດທ້າຍແລະສືບຕໍ່ການຄິດໄລ່ຈາກມັນເພື່ອຟື້ນຕົວຂໍ້ມູນສູນເສຍ.

ການກວດສອບສາມາດຖືກເປີດໃຊ້ໂດຍການຕັ້ງຄ່າໄດເລກະທໍລີໃນລະບົບໄຟລ໌ທີ່ທົນທານຕໍ່ຄວາມຜິດ, ເຊື່ອຖືໄດ້ (ເຊັ່ນ HDFS, S3, ແລະອື່ນໆ) ເຊິ່ງຂໍ້ມູນດ່ານຈະຖືກເກັບໄວ້. ນີ້ແມ່ນເຮັດໄດ້ໂດຍໃຊ້ຕົວຢ່າງ:

streamingContext.checkpoint(checkpointDirectory)

ໃນຕົວຢ່າງຂອງພວກເຮົາ, ພວກເຮົາຈະນໍາໃຊ້ວິທີການດັ່ງຕໍ່ໄປນີ້, ຄື, ຖ້າ checkpointDirectory ມີ, ຫຼັງຈາກນັ້ນສະພາບການຈະຖືກສ້າງໃຫມ່ຈາກຂໍ້ມູນດ່ານ. ຖ້າໄດເລກະທໍລີບໍ່ມີ (i. e. ປະຕິບັດຄັ້ງທໍາອິດ), ຫຼັງຈາກນັ້ນ 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, ພວກເຮົາເຮັດການຈັດກຸ່ມແບບງ່າຍດາຍແລະສະແດງຜົນໄດ້ຮັບໃນ 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

ເອົາຂໍ້ຄວາມສອບຖາມແລະແລ່ນມັນຜ່ານ 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. ພວກເຮົາສ້າງຜູ້ໃຊ້ແລະລະຫັດຜ່ານສໍາລັບມັນຢູ່ໃນຂັ້ນຕອນ "Deploying AWS PostgreSQL". ທ່ານຄວນໃຊ້ Endpoint ເປັນ url ເຊີບເວີຖານຂໍ້ມູນ, ເຊິ່ງສະແດງຢູ່ໃນພາກ Connectivity & security:

ເພື່ອເຊື່ອມຕໍ່ Spark ແລະ Kafka ຢ່າງຖືກຕ້ອງ, ທ່ານຄວນເຮັດວຽກຜ່ານ smark-submit ໂດຍໃຊ້ artifact. 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 ວິນາທີ, ເພາະວ່າພວກເຮົາກໍານົດຊ່ວງເວລາ batching ເປັນ 2 ວິນາທີເມື່ອພວກເຮົາສ້າງວັດຖຸ StreamingContext:

ຕໍ່ໄປ, ພວກເຮົາເຮັດແບບສອບຖາມງ່າຍໆກັບຖານຂໍ້ມູນເພື່ອກວດເບິ່ງການປະກົດຕົວຂອງບັນທຶກໃນຕາຕະລາງ transaction_flow:

ສະຫລຸບ

ບົດຄວາມນີ້ໄດ້ເບິ່ງຕົວຢ່າງຂອງການປະມວນຜົນຂໍ້ມູນໂດຍໃຊ້ Spark Streaming ໂດຍສົມທົບກັບ Apache Kafka ແລະ PostgreSQL. ດ້ວຍການຂະຫຍາຍຕົວຂອງຂໍ້ມູນຈາກແຫຼ່ງຕ່າງໆ, ມັນເປັນການຍາກທີ່ຈະປະເມີນມູນຄ່າການປະຕິບັດຂອງ Spark Streaming ເກີນໄປສໍາລັບການສ້າງການຖ່າຍທອດແລະຄໍາຮ້ອງສະຫມັກໃນເວລາຈິງ.

ທ່ານສາມາດຊອກຫາລະຫັດແຫຼ່ງເຕັມໃນ repository ຂອງຂ້ອຍຢູ່ທີ່ GitHub.

ຂ້າພະເຈົ້າຍິນດີທີ່ຈະສົນທະນາບົດຄວາມນີ້, ຂ້າພະເຈົ້າຫວັງວ່າຈະໄດ້ຄໍາຄິດເຫັນຂອງທ່ານ, ແລະຂ້າພະເຈົ້າຍັງຫວັງວ່າສໍາລັບການວິພາກວິຈານການກໍ່ສ້າງຈາກຜູ້ອ່ານທີ່ເປັນຫ່ວງເປັນໄຍທຸກຄົນ.

ຂ້າ​ພະ​ເຈົ້າ​ຂໍ​ໃຫ້​ທ່ານ​ສົບ​ຜົນ​ສໍາ​ເລັດ​!

ປ. ໃນເບື້ອງຕົ້ນມັນໄດ້ຖືກວາງແຜນທີ່ຈະໃຊ້ຖານຂໍ້ມູນ PostgreSQL ທ້ອງຖິ່ນ, ແຕ່ຍ້ອນຄວາມຮັກຂອງຂ້ອຍສໍາລັບ AWS, ຂ້ອຍໄດ້ຕັດສິນໃຈຍ້າຍຖານຂໍ້ມູນໄປຍັງເມຄ. ໃນບົດຄວາມຕໍ່ໄປກ່ຽວກັບຫົວຂໍ້ນີ້, ຂ້າພະເຈົ້າຈະສະແດງວິທີການປະຕິບັດລະບົບທັງຫມົດທີ່ໄດ້ອະທິບາຍຂ້າງເທິງໃນ AWS ໂດຍໃຊ້ AWS Kinesis ແລະ AWS EMR. ຕິດຕາມຂ່າວ!

ແຫຼ່ງຂໍ້ມູນ: www.habr.com