স্পার্ক স্ট্রিমিং সহ অ্যাপাচি কাফকা এবং ডেটা স্ট্রিমিং

হে হাবর! আজ আমরা একটি সিস্টেম তৈরি করব যা স্পার্ক স্ট্রিমিং ব্যবহার করে Apache Kafka বার্তা স্ট্রীমগুলিকে প্রক্রিয়া করবে এবং প্রক্রিয়াকরণের ফলাফল AWS RDS ক্লাউড ডাটাবেসে লিখবে৷

আসুন কল্পনা করি যে একটি নির্দিষ্ট ক্রেডিট প্রতিষ্ঠান আমাদের সামনে তার সমস্ত শাখার জন্য "ফ্লাই" ইনকামিং লেনদেন প্রক্রিয়াকরণের কাজটি সেট করে। এটি কোষাগারের জন্য খোলা মুদ্রার অবস্থান, লেনদেনের সীমা বা আর্থিক ফলাফল ইত্যাদির দ্রুত গণনা করার উদ্দেশ্যে করা যেতে পারে।

জাদু এবং জাদু মন্ত্র ব্যবহার ছাড়া এই ক্ষেত্রে কিভাবে বাস্তবায়ন - কাটা অধীনে পড়ুন! যাওয়া!

(ছবির সূত্র)

ভূমিকা

অবশ্যই, প্রচুর পরিমাণে ডেটার রিয়েল-টাইম প্রক্রিয়াকরণ আধুনিক সিস্টেমে ব্যবহারের জন্য যথেষ্ট সুযোগ প্রদান করে। এর জন্য সবচেয়ে জনপ্রিয় সমন্বয়গুলির মধ্যে একটি হল অ্যাপাচি কাফকা এবং স্পার্ক স্ট্রিমিং এর টেন্ডেম, যেখানে কাফকা ইনকামিং বার্তা প্যাকেটগুলির একটি স্ট্রীম তৈরি করে এবং স্পার্ক স্ট্রিমিং একটি নির্দিষ্ট সময়ের ব্যবধানে এই প্যাকেটগুলিকে প্রক্রিয়া করে।

অ্যাপ্লিকেশনের ত্রুটি সহনশীলতা উন্নত করতে, আমরা চেকপয়েন্ট - চেকপয়েন্ট ব্যবহার করব। এই পদ্ধতির সাহায্যে, যখন স্পার্ক স্ট্রিমিং মডিউলটি হারিয়ে যাওয়া ডেটা পুনরুদ্ধার করতে হবে, তখন এটি শুধুমাত্র শেষ চেকপয়েন্টে ফিরে যেতে হবে এবং সেখান থেকে গণনা আবার শুরু করতে হবে।

উন্নত সিস্টেমের আর্কিটেকচার

ব্যবহৃত উপাদান:

• আপাচি কাফকা একটি বিতরণ করা প্রকাশ এবং সদস্যতা বার্তা সিস্টেম. অফলাইন এবং অনলাইন উভয় বার্তা ব্যবহারের জন্য উপযুক্ত। ডেটা ক্ষতি রোধ করতে, কাফকা বার্তাগুলি ডিস্কে সংরক্ষণ করা হয় এবং ক্লাস্টারের মধ্যে প্রতিলিপি করা হয়। কাফকা সিস্টেম জুকিপার সিঙ্ক্রোনাইজেশন পরিষেবার উপরে নির্মিত;
• অ্যাপাচি স্পার্ক স্ট্রিমিং - স্ট্রিমিং ডেটা প্রক্রিয়াকরণের জন্য একটি স্পার্ক উপাদান। স্পার্ক স্ট্রিমিং মডিউলটি একটি "মাইক্রো-ব্যাচ" আর্কিটেকচার ব্যবহার করে তৈরি করা হয়, যখন একটি ডেটা স্ট্রিমকে ছোট ডেটা প্যাকেটগুলির একটি ক্রমাগত ক্রম হিসাবে ব্যাখ্যা করা হয়। স্পার্ক স্ট্রিমিং বিভিন্ন উত্স থেকে ডেটা নেয় এবং এটিকে ছোট ব্যাচে একত্রিত করে। নিয়মিত বিরতিতে নতুন প্যাকেজ তৈরি করা হয়। প্রতিটি সময়ের ব্যবধানের শুরুতে, একটি নতুন প্যাকেট তৈরি করা হয় এবং সেই ব্যবধানে প্রাপ্ত যেকোন ডেটা প্যাকেটে অন্তর্ভুক্ত করা হয়। ব্যবধানের শেষে, প্যাকেটের বৃদ্ধি বন্ধ হয়ে যায়। ব্যাচ ব্যবধান নামক একটি প্যারামিটার দ্বারা ব্যবধানের আকার নির্ধারণ করা হয়;
• অ্যাপাচি স্পার্ক এসকিউএল - স্পার্ক ফাংশনাল প্রোগ্রামিংয়ের সাথে রিলেশনাল প্রসেসিংকে একত্রিত করে। স্ট্রাকচার্ড ডেটা বলতে এমন ডেটা বোঝায় যার একটি স্কিমা আছে, অর্থাৎ সমস্ত রেকর্ডের জন্য একটি একক ফিল্ড। স্পার্ক এসকিউএল বিভিন্ন স্ট্রাকচার্ড ডেটা সোর্স থেকে ইনপুট সমর্থন করে এবং স্কিমা তথ্যের উপস্থিতির কারণে, এটি দক্ষতার সাথে শুধুমাত্র প্রয়োজনীয় ক্ষেত্রগুলি পুনরুদ্ধার করতে পারে এবং ডেটাফ্রেম এপিআইও প্রদান করে;
• 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

নতুন তৈরি বিষয়ের জন্য প্রযোজক এবং ভোক্তাদের পরীক্ষার মুহূর্তগুলি মিস করা যাক। আপনি কীভাবে বার্তা পাঠানো এবং গ্রহণ করা পরীক্ষা করতে পারেন তার আরও বিশদ বিবরণ অফিসিয়াল ডকুমেন্টেশনে লেখা আছে - কিছু বার্তা পাঠান. ঠিক আছে, আমরা 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()

স্ক্রিপ্ট চালানোর সময়, আমরা টার্মিনালে নিম্নলিখিত বার্তাগুলি পাই:

এর মানে হল যে সবকিছু আমরা যেমন চেয়েছিলাম তেমন কাজ করে - প্রযোজক আমাদের প্রয়োজনীয় বিষয়ে বার্তা তৈরি করে এবং পাঠায়।
পরবর্তী ধাপ হল স্পার্ক ইনস্টল করা এবং এই বার্তা প্রবাহ প্রক্রিয়া করা।

অ্যাপাচি স্পার্ক ইনস্টল করা হচ্ছে

আপা স্পার্ক একটি বহুমুখী এবং উচ্চ কর্মক্ষমতা ক্লাস্টার কম্পিউটিং প্ল্যাটফর্ম।

পারফরম্যান্সের ক্ষেত্রে স্পার্ক ম্যাপরিডুস মডেলের জনপ্রিয় বাস্তবায়নকে ছাড়িয়ে যায়, যেখানে ইন্টারেক্টিভ কোয়েরি এবং স্ট্রিমিং সহ গণনার প্রকারের বিস্তৃত পরিসরের জন্য সমর্থন প্রদান করে। প্রচুর পরিমাণে ডেটা প্রক্রিয়া করার সময় গতি একটি গুরুত্বপূর্ণ ভূমিকা পালন করে, যেহেতু এটি গতি যা আপনাকে মিনিট বা ঘন্টা অপেক্ষা না করে ইন্টারেক্টিভভাবে কাজ করতে দেয়। এই গতি প্রদানের জন্য স্পার্কের সবচেয়ে বড় শক্তি হল এটির ইন-মেমরি গণনা করার ক্ষমতা।

এই ফ্রেমওয়ার্কটি স্কালায় লেখা হয়েছে, তাই আপনাকে প্রথমে এটি ইনস্টল করতে হবে:

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 ক্লাসের একটি উদাহরণ দেওয়া হবে - যদিও দুর্বল, কিন্তু বিনামূল্যে এবং আমাদের কাজের জন্য বেশ উপযুক্ত:

খুব গুরুত্বপূর্ণ বিষয়গুলি অনুসরণ করে: DB উদাহরণের নাম, মাস্টার ব্যবহারকারীর নাম এবং তার পাসওয়ার্ড৷ আসুন উদাহরণটিকে কল করি: myHabrTest, মাস্টার ব্যবহারকারী: হাবর, পাসওয়ার্ড: habr12345 এবং পরবর্তী বোতামে ক্লিক করুন:

পরের পৃষ্ঠায় আমাদের ডাটাবেস সার্ভারের বাইরে থেকে অ্যাক্সেসযোগ্যতা (পাবলিক অ্যাক্সেসিবিলিটি) এবং পোর্টগুলির উপলব্ধতার জন্য দায়ী পরামিতিগুলি রয়েছে:

আসুন VPC নিরাপত্তা গোষ্ঠীর জন্য একটি নতুন সেটিং তৈরি করি, যা পোর্ট 5432 (PostgreSQL) এর মাধ্যমে আমাদের ডাটাবেস সার্ভারে বাহ্যিক অ্যাক্সেসের অনুমতি দেবে।
VPC ড্যাশবোর্ড -> সিকিউরিটি গ্রুপ -> সিকিউরিটি গ্রুপ সেকশনে AWS কনসোলে একটি আলাদা ব্রাউজার উইন্ডোতে যাওয়া যাক:

আমরা নিরাপত্তা গোষ্ঠীর জন্য নাম সেট করেছি - পোস্টগ্রেএসকিউএল, একটি বিবরণ, এই গ্রুপটি কোন ভিপিসির সাথে যুক্ত হবে তা নির্দিষ্ট করুন এবং তৈরি করুন বোতামটি ক্লিক করুন:

আমরা পোর্ট 5432-এর জন্য নতুন তৈরি করা গ্রুপ ইনবাউন্ড নিয়মগুলি পূরণ করি, যেমনটি নীচের ছবিতে দেখানো হয়েছে। আপনি ম্যানুয়ালি পোর্ট নির্দিষ্ট করতে পারবেন না, তবে টাইপ ড্রপ-ডাউন তালিকা থেকে PostgreSQL নির্বাচন করুন।

কঠোরভাবে বলতে গেলে, মান ::/0 মানে সারা বিশ্ব থেকে সার্ভারের জন্য ইনকামিং ট্র্যাফিকের প্রাপ্যতা, যা ক্যানোনিকভাবে পুরোপুরি সত্য নয়, কিন্তু উদাহরণটি বিশ্লেষণ করতে, আসুন এই পদ্ধতিটি ব্যবহার করা যাক:

আমরা ব্রাউজার পৃষ্ঠায় ফিরে আসি, যেখানে আমাদের "উন্নত সেটিংস কনফিগার করুন" খোলা আছে এবং VPC নিরাপত্তা গোষ্ঠী নির্বাচন করুন -> বিদ্যমান VPC নিরাপত্তা গোষ্ঠীগুলি চয়ন করুন -> বিভাগে পোস্টগ্রেএসকিউএল:

এরপরে, বিভাগে ডেটাবেস বিকল্প -> ডেটাবেস নাম -> নাম সেট করুন - habrDB.

আমরা ডিফল্টরূপে ব্যাকআপ (ব্যাকআপ ধরে রাখার সময়কাল - 0 দিন), পর্যবেক্ষণ এবং পারফরম্যান্স অন্তর্দৃষ্টি অক্ষম করা বাদে বাকি প্যারামিটারগুলি ছেড়ে দিতে পারি। বোতামে ক্লিক করুন ডাটাবেস তৈরি করুন:

স্ট্রিম হ্যান্ডলার

চূড়ান্ত পর্যায়ে একটি স্পার্ক কাজের বিকাশ হবে, যা প্রতি দুই সেকেন্ডে কাফকা থেকে নতুন ডেটা প্রক্রিয়া করবে এবং ফলাফল ডাটাবেসে প্রবেশ করবে।

উপরে উল্লিখিত হিসাবে, চেকপয়েন্টগুলি হল স্পার্কস্ট্রিমিং-এর প্রধান প্রক্রিয়া যা ত্রুটি সহনশীলতা প্রদানের জন্য কনফিগার করা আবশ্যক। আমরা চেকপয়েন্ট ব্যবহার করব এবং, একটি পদ্ধতির ব্যর্থতার ক্ষেত্রে, স্পার্ক স্ট্রিমিং মডিউলটিকে শুধুমাত্র শেষ চেকপয়েন্টে ফিরে যেতে হবে এবং হারিয়ে যাওয়া ডেটা পুনরুদ্ধার করতে এটি থেকে গণনা পুনরায় শুরু করতে হবে।

একটি ত্রুটি-সহনশীল, নির্ভরযোগ্য ফাইল সিস্টেমে (যেমন HDFS, S3, ইত্যাদি) একটি ডিরেক্টরি সেট করে একটি চেকপয়েন্ট সক্রিয় করা যেতে পারে যেখানে চেকপয়েন্ট তথ্য সংরক্ষণ করা হবে। এটি করা হয়, উদাহরণস্বরূপ:

streamingContext.checkpoint(checkpointDirectory)

আমাদের উদাহরণে, আমরা নিম্নলিখিত পদ্ধতি ব্যবহার করব, যথা, যদি চেকপয়েন্ট ডাইরেক্টরি বিদ্যমান থাকে, তাহলে চেকপয়েন্ট ডেটা থেকে প্রসঙ্গটি পুনরায় তৈরি করা হবে। যদি ডিরেক্টরিটি বিদ্যমান না থাকে (অর্থাৎ এটি প্রথমবারের মতো কার্যকর করা হচ্ছে), তাহলে একটি নতুন প্রসঙ্গ তৈরি করতে এবং DStreams সেট আপ করার জন্য 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")

স্পার্ক এসকিউএল ব্যবহার করে, আমরা একটি সাধারণ গ্রুপিং করি এবং ফলাফলটি কনসোলে আউটপুট করি:

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 স্থাপন" ধাপে এটির জন্য ব্যবহারকারী এবং পাসওয়ার্ড তৈরি করেছি। ডাটাবেস সার্ভারের ইউআরএল হিসাবে, আপনার এন্ডপয়েন্ট ব্যবহার করা উচিত, যা সংযোগ এবং নিরাপত্তা বিভাগে প্রদর্শিত হয়:

স্পার্ক এবং কাফকাকে সঠিকভাবে সংযোগ করার জন্য, আপনাকে আর্টিফ্যাক্ট ব্যবহার করে 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 সেকেন্ডে নতুন একত্রিত ফলাফল প্রদর্শিত হয়, কারণ আমরা যখন StreamingContext অবজেক্ট তৈরি করি তখন আমরা বান্ডলিং ব্যবধান 2 সেকেন্ডে সেট করি:

এর পরে, আমরা টেবিলে রেকর্ডগুলি পরীক্ষা করার জন্য ডাটাবেসের কাছে একটি সাধারণ প্রশ্ন করি লেনদেন_প্রবাহ:

উপসংহার

এই নিবন্ধে, Apache Kafka এবং PostgreSQL এর সাথে স্পার্ক স্ট্রিমিং ব্যবহার করে স্ট্রিমিং তথ্য প্রক্রিয়াকরণের একটি উদাহরণ বিবেচনা করা হয়েছিল। বিভিন্ন উত্স থেকে ডেটার পরিমাণ বৃদ্ধির সাথে, রিয়েল-টাইম এবং স্ট্রিমিং অ্যাপ্লিকেশন তৈরির জন্য স্পার্ক স্ট্রিমিংয়ের ব্যবহারিক মূল্যকে অত্যধিক মূল্যায়ন করা কঠিন।

আপনি আমার সংগ্রহস্থলে সম্পূর্ণ উৎস কোড খুঁজে পেতে পারেন GitHub.

আমি এই নিবন্ধটি নিয়ে আলোচনা করতে পেরে আনন্দিত, আমি আপনার মন্তব্যের জন্য অপেক্ষা করছি, এবং এছাড়াও, আমি সমস্ত সংশ্লিষ্ট পাঠকদের কাছ থেকে গঠনমূলক সমালোচনা আশা করছি।

আমি আপনাকে সাফল্য কামনা করি!

গীত। এটি মূলত একটি স্থানীয় PostgreSQL ডাটাবেস ব্যবহার করার পরিকল্পনা করা হয়েছিল, কিন্তু AWS এর প্রতি আমার ভালবাসার কারণে, আমি ডাটাবেসটিকে ক্লাউডে সরানোর সিদ্ধান্ত নিয়েছি। এই বিষয়ে পরবর্তী নিবন্ধে, আমি আপনাকে AWS Kinesis এবং AWS EMR ব্যবহার করে AWS-এ উপরে বর্ণিত সম্পূর্ণ সিস্টেমটি কীভাবে বাস্তবায়ন করতে হয় তা দেখাব। খবর অনুসরণ করুন!

উত্স: www.habr.com