Amazon Kinesis এবং সার্ভারহীন সরলতার সাথে Aviasales API ইন্টিগ্রেশন

হে হাবর!

আপনি কি উড়ন্ত বিমান পছন্দ করেন? আমি এটি পছন্দ করি, তবে স্ব-বিচ্ছিন্নতার সময় আমি একটি সুপরিচিত সংস্থান - অ্যাভিয়াসেলস থেকে এয়ার টিকিটের ডেটা বিশ্লেষণের প্রেমে পড়েছিলাম।

আজ আমরা Amazon Kinesis-এর কাজ বিশ্লেষণ করব, রিয়েল-টাইম অ্যানালিটিক্স সহ একটি স্ট্রিমিং সিস্টেম তৈরি করব, Amazon DynamoDB NoSQL ডাটাবেসকে প্রধান ডেটা স্টোরেজ হিসাবে ইনস্টল করব এবং আকর্ষণীয় টিকিটের জন্য SMS বিজ্ঞপ্তিগুলি সেট আপ করব৷

সব বিবরণ কাটা অধীনে! যাওয়া!

ভূমিকা

উদাহরণস্বরূপ, আমাদের অ্যাক্সেস দরকার Aviasales API. এটিতে অ্যাক্সেস বিনামূল্যে এবং বিধিনিষেধ ছাড়াই সরবরাহ করা হয়; ডেটা অ্যাক্সেস করতে আপনার API টোকেন পেতে আপনাকে কেবল "ডেভেলপার" বিভাগে নিবন্ধন করতে হবে৷

এই নিবন্ধটির মূল লক্ষ্য হল AWS-এ তথ্য স্ট্রিমিং ব্যবহার সম্পর্কে একটি সাধারণ ধারণা প্রদান করা; আমরা বিবেচনা করি যে API দ্বারা প্রত্যাবর্তিত ডেটা কঠোরভাবে আপ-টু-ডেট নয় এবং ক্যাশে থেকে স্থানান্তরিত হয়, যা গত 48 ঘন্টা ধরে Aviasales.ru এবং Jetradar.com সাইটের ব্যবহারকারীদের অনুসন্ধানের ভিত্তিতে গঠিত।

কাইনেসিস-এজেন্ট, উত্পাদনকারী মেশিনে ইনস্টল করা, API-এর মাধ্যমে প্রাপ্ত, স্বয়ংক্রিয়ভাবে কাইনেসিস ডেটা অ্যানালিটিক্সের মাধ্যমে পছন্দসই স্ট্রীমে ডেটা পার্স এবং প্রেরণ করবে। এই স্ট্রীমের কাঁচা সংস্করণটি সরাসরি দোকানে লেখা হবে। DynamoDB-তে স্থাপন করা কাঁচা ডেটা স্টোরেজ BI টুলের মাধ্যমে গভীর টিকিট বিশ্লেষণের অনুমতি দেবে, যেমন AWS Quick Sight।

আমরা সম্পূর্ণ অবকাঠামো স্থাপনের জন্য দুটি বিকল্প বিবেচনা করব:

• ম্যানুয়াল - AWS ম্যানেজমেন্ট কনসোলের মাধ্যমে;
• Terraform কোড থেকে অবকাঠামো অলস অটোমেটরদের জন্য;

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

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

• Aviasales API — এই API দ্বারা প্রত্যাবর্তিত ডেটা পরবর্তী সমস্ত কাজের জন্য ব্যবহার করা হবে;
• EC2 প্রযোজক উদাহরণ — ক্লাউডে একটি নিয়মিত ভার্চুয়াল মেশিন যার উপর ইনপুট ডেটা স্ট্রিম তৈরি করা হবে:
  • কাইনেসিস এজেন্ট মেশিনে স্থানীয়ভাবে ইনস্টল করা একটি জাভা অ্যাপ্লিকেশন যা কাইনেসিস (কাইনেসিস ডেটা স্ট্রিম বা কাইনেসিস ফায়ারহোস) ডেটা সংগ্রহ ও পাঠানোর একটি সহজ উপায় প্রদান করে। এজেন্ট ক্রমাগত নির্দিষ্ট ডিরেক্টরিতে ফাইলগুলির একটি সেট নিরীক্ষণ করে এবং কাইনেসিসে নতুন ডেটা পাঠায়;
  • API কলার স্ক্রিপ্ট — একটি পাইথন স্ক্রিপ্ট যা API-কে অনুরোধ করে এবং প্রতিক্রিয়াটিকে একটি ফোল্ডারে রাখে যা কাইনেসিস এজেন্ট দ্বারা পর্যবেক্ষণ করা হয়;
• কাইনেসিস ডেটা স্ট্রীম - ব্যাপক স্কেলিং ক্ষমতা সহ রিয়েল-টাইম ডেটা স্ট্রিমিং পরিষেবা;
• কাইনেসিস অ্যানালিটিক্স একটি সার্ভারহীন পরিষেবা যা রিয়েল টাইমে স্ট্রিমিং ডেটা বিশ্লেষণকে সহজ করে। অ্যামাজন কাইনেসিস ডেটা অ্যানালিটিক্স অ্যাপ্লিকেশন সংস্থানগুলি কনফিগার করে এবং স্বয়ংক্রিয়ভাবে আগত ডেটার যে কোনও ভলিউম পরিচালনা করতে স্কেল করে;
• এডাব্লুএস ল্যাম্বদা — এমন একটি পরিষেবা যা আপনাকে সার্ভার ব্যাক আপ বা সেট আপ না করেই কোড চালানোর অনুমতি দেয়। প্রতিটি কলের জন্য সমস্ত কম্পিউটিং শক্তি স্বয়ংক্রিয়ভাবে স্কেল করা হয়;
• আমাজন ডায়নামোডিবি - কী-মানের জোড়া এবং নথিগুলির একটি ডাটাবেস যা যেকোনো স্কেলে চলাকালীন 10 মিলিসেকেন্ডের কম বিলম্ব প্রদান করে। DynamoDB ব্যবহার করার সময়, আপনাকে কোনো সার্ভারের ব্যবস্থা, প্যাচ বা পরিচালনা করতে হবে না। DynamoDB স্বয়ংক্রিয়ভাবে উপলব্ধ সংস্থানগুলির পরিমাণ সামঞ্জস্য করতে এবং উচ্চ কার্যক্ষমতা বজায় রাখতে টেবিলগুলিকে স্কেল করে। কোন সিস্টেম প্রশাসনের প্রয়োজন নেই;
• অ্যামাজন এসএনএস - প্রকাশক-সাবস্ক্রাইবার (পাব/সাব) মডেল ব্যবহার করে বার্তা পাঠানোর জন্য একটি সম্পূর্ণরূপে পরিচালিত পরিষেবা, যার সাহায্যে আপনি মাইক্রোসার্ভিস, বিতরণ করা সিস্টেম এবং সার্ভারহীন অ্যাপ্লিকেশনগুলিকে আলাদা করতে পারেন৷ মোবাইল পুশ বিজ্ঞপ্তি, এসএমএস বার্তা এবং ইমেলের মাধ্যমে শেষ ব্যবহারকারীদের কাছে তথ্য পাঠাতে SNS ব্যবহার করা যেতে পারে।

প্রাথমিক প্রশিক্ষণ

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

সুতরাং, আসুন নিবন্ধন করুন এবং আমাদের টোকেন পান।

একটি উদাহরণ অনুরোধ নীচে দেওয়া হল:

http://api.travelpayouts.com/v2/prices/month-matrix?currency=rub&origin=LED&destination=HKT&show_to_affiliates=true&token=TOKEN_API

অনুরোধে একটি টোকেন উল্লেখ করে API থেকে ডেটা গ্রহণের উপরের পদ্ধতিটি কাজ করবে, তবে আমি হেডারের মাধ্যমে অ্যাক্সেস টোকেন পাস করতে পছন্দ করি, তাই আমরা api_caller.py স্ক্রিপ্টে এই পদ্ধতিটি ব্যবহার করব।

উত্তর উদাহরণ:

{{
   "success":true,
   "data":[{
      "show_to_affiliates":true,
      "trip_class":0,
      "origin":"LED",
      "destination":"HKT",
      "depart_date":"2015-10-01",
      "return_date":"",
      "number_of_changes":1,
      "value":29127,
      "found_at":"2015-09-24T00:06:12+04:00",
      "distance":8015,
      "actual":true
   }]
}

উপরের উদাহরণ API প্রতিক্রিয়াটি সেন্ট পিটার্সবার্গ থেকে ফুক পর্যন্ত একটি টিকিট দেখায়... ওহ, কী স্বপ্ন...
যেহেতু আমি কাজান থেকে এসেছি, এবং ফুকেট এখন "শুধুমাত্র একটি স্বপ্ন", আসুন সেন্ট পিটার্সবার্গ থেকে কাজান পর্যন্ত টিকিট খোঁজা যাক।

এটি অনুমান করে যে আপনার ইতিমধ্যে একটি AWS অ্যাকাউন্ট আছে। আমি অবিলম্বে এই বিষয়টির প্রতি বিশেষ দৃষ্টি আকর্ষণ করতে চাই যে কাইনেসিস এবং এসএমএসের মাধ্যমে বিজ্ঞপ্তি পাঠানো বার্ষিক অন্তর্ভুক্ত নয় বিনামূল্যের স্তর (বিনামূল্যে ব্যবহার). তবে তা সত্ত্বেও, কয়েক ডলার মাথায় রেখে, প্রস্তাবিত সিস্টেমটি তৈরি করা এবং এটির সাথে খেলা বেশ সম্ভব। এবং, অবশ্যই, সমস্ত সম্পদ মুছে ফেলতে ভুলবেন না পরে তাদের আর প্রয়োজন নেই।

সৌভাগ্যবশত, DynamoDb এবং lambda ফাংশন আমাদের জন্য বিনামূল্যে হবে যদি আমরা আমাদের মাসিক বিনামূল্যের সীমা পূরণ করি। উদাহরণস্বরূপ, DynamoDB এর জন্য: 25 GB স্টোরেজ, 25 WCU/RCU এবং 100 মিলিয়ন প্রশ্ন। এবং প্রতি মাসে এক মিলিয়ন ল্যাম্বডা ফাংশন কল।

ম্যানুয়াল সিস্টেম স্থাপনা

কাইনেসিস ডেটা স্ট্রীম সেট আপ করা হচ্ছে

আসুন কাইনেসিস ডেটা স্ট্রিম পরিষেবাতে যাই এবং দুটি নতুন স্ট্রিম তৈরি করি, প্রতিটির জন্য একটি শার্ড।

একটি শার্ড কি?
একটি শার্ড হল একটি অ্যামাজন কাইনেসিস স্ট্রিমের মৌলিক ডেটা স্থানান্তর ইউনিট। একটি সেগমেন্ট 1 MB/s গতিতে ইনপুট ডেটা স্থানান্তর এবং 2 MB/s গতিতে আউটপুট ডেটা স্থানান্তর প্রদান করে। একটি সেগমেন্ট প্রতি সেকেন্ডে 1000টি PUT এন্ট্রি সমর্থন করে। একটি ডেটা স্ট্রিম তৈরি করার সময়, আপনাকে প্রয়োজনীয় সংখ্যক সেগমেন্ট উল্লেখ করতে হবে। উদাহরণস্বরূপ, আপনি দুটি বিভাগ সহ একটি ডেটা স্ট্রিম তৈরি করতে পারেন। এই ডেটা স্ট্রীমটি 2 MB/s গতিতে ইনপুট ডেটা স্থানান্তর এবং 4 MB/s এ আউটপুট ডেটা স্থানান্তর প্রদান করবে, প্রতি সেকেন্ডে 2000 PUT রেকর্ডগুলিকে সমর্থন করবে৷

আপনার স্ট্রীমে যত বেশি শার্ড, তার থ্রুপুট তত বেশি। নীতিগতভাবে, এইভাবে প্রবাহকে মাপানো হয় - শার্ড যোগ করে। কিন্তু আপনার যত বেশি শার্ড থাকবে, দাম তত বেশি হবে। প্রতিটি শার্ডের দাম প্রতি ঘন্টায় 1,5 সেন্ট এবং প্রতি মিলিয়ন PUT পেলোড ইউনিটের জন্য অতিরিক্ত 1.4 সেন্ট।

নাম দিয়ে একটি নতুন স্ট্রিম তৈরি করা যাক বিমান টিকেট, 1 শার্ড তার জন্য যথেষ্ট হবে:

এখন নাম দিয়ে আরেকটি থ্রেড তৈরি করা যাক বিশেষ_প্রবাহ:

প্রযোজক সেটআপ

একটি টাস্ক বিশ্লেষণ করার জন্য, ডেটা প্রযোজক হিসাবে একটি নিয়মিত EC2 উদাহরণ ব্যবহার করা যথেষ্ট। এটি একটি শক্তিশালী, ব্যয়বহুল ভার্চুয়াল মেশিন হতে হবে না; একটি স্পট t2.micro ঠিক কাজ করবে।

গুরুত্বপূর্ণ দ্রষ্টব্য: উদাহরণস্বরূপ, আপনার ইমেজ ব্যবহার করা উচিত - Amazon Linux AMI 2018.03.0, এটিতে দ্রুত Kinesis এজেন্ট চালু করার জন্য কম সেটিংস রয়েছে।

EC2 পরিষেবাতে যান, একটি নতুন ভার্চুয়াল মেশিন তৈরি করুন, t2.micro টাইপ সহ পছন্দসই AMI নির্বাচন করুন, যা বিনামূল্যের স্তরে অন্তর্ভুক্ত:

নতুন তৈরি ভার্চুয়াল মেশিনটি কাইনেসিস পরিষেবার সাথে ইন্টারঅ্যাক্ট করতে সক্ষম হওয়ার জন্য, এটি করার অধিকার অবশ্যই দেওয়া উচিত। এটি করার সর্বোত্তম উপায় হল একটি IAM ভূমিকা বরাদ্দ করা। অতএব, ধাপ 3: কনফিগার ইনস্ট্যান্স বিবরণ স্ক্রিনে, আপনার নির্বাচন করা উচিত নতুন IAM ভূমিকা তৈরি করুন:

EC2 এর জন্য একটি IAM ভূমিকা তৈরি করা

যে উইন্ডোটি খোলে, সেখানে নির্বাচন করুন যে আমরা EC2 এর জন্য একটি নতুন ভূমিকা তৈরি করছি এবং অনুমতি বিভাগে যান:

প্রশিক্ষণের উদাহরণ ব্যবহার করে, আমাদের সম্পদ অধিকারের দানাদার কনফিগারেশনের সমস্ত জটিলতার মধ্যে যেতে হবে না, তাই আমরা Amazon দ্বারা পূর্ব-কনফিগার করা নীতিগুলি নির্বাচন করব: AmazonKinesisFullAccess এবং CloudWatchFullAccess।

এই ভূমিকার কিছু অর্থপূর্ণ নাম দেওয়া যাক, উদাহরণস্বরূপ: EC2-KinesisStreams-FullAccess. ফলাফল নীচের ছবিতে দেখানো হিসাবে একই হওয়া উচিত:

এই নতুন ভূমিকা তৈরি করার পরে, তৈরি করা ভার্চুয়াল মেশিনের উদাহরণে এটি সংযুক্ত করতে ভুলবেন না:

আমরা এই স্ক্রিনে অন্য কিছু পরিবর্তন করি না এবং পরবর্তী উইন্ডোতে চলে যাই।

হার্ড ড্রাইভ সেটিংস ডিফল্ট হিসাবে রেখে দেওয়া যেতে পারে, সেইসাথে ট্যাগগুলি (যদিও ট্যাগগুলি ব্যবহার করা ভাল অভ্যাস, অন্তত উদাহরণটিকে একটি নাম দিন এবং পরিবেশ নির্দেশ করুন)৷

এখন আমরা স্টেপ 6: কনফিগার সিকিউরিটি গ্রুপ ট্যাবে আছি, যেখানে আপনাকে একটি নতুন তৈরি করতে হবে বা আপনার বিদ্যমান সিকিউরিটি গ্রুপ নির্দিষ্ট করতে হবে, যা আপনাকে ssh (পোর্ট 22) এর মাধ্যমে ইনস্ট্যান্সে সংযোগ করতে দেয়। উত্স -> সেখানে আমার আইপি নির্বাচন করুন এবং আপনি উদাহরণটি চালু করতে পারেন।

এটি চলমান স্থিতিতে স্যুইচ করার সাথে সাথে আপনি ssh এর মাধ্যমে এটির সাথে সংযোগ করার চেষ্টা করতে পারেন।

কাইনেসিস এজেন্টের সাথে কাজ করতে সক্ষম হওয়ার জন্য, মেশিনে সফলভাবে সংযোগ করার পরে, আপনাকে অবশ্যই টার্মিনালে নিম্নলিখিত কমান্ডগুলি প্রবেশ করতে হবে:

sudo yum -y update
sudo yum install -y python36 python36-pip
sudo /usr/bin/pip-3.6 install --upgrade pip
sudo yum install -y aws-kinesis-agent

API প্রতিক্রিয়া সংরক্ষণ করতে একটি ফোল্ডার তৈরি করা যাক:

sudo mkdir /var/log/airline_tickets

এজেন্ট শুরু করার আগে, আপনাকে এর কনফিগারেশন কনফিগার করতে হবে:

sudo vim /etc/aws-kinesis/agent.json

agent.json ফাইলের বিষয়বস্তু এইরকম হওয়া উচিত:

{
  "cloudwatch.emitMetrics": true,
  "kinesis.endpoint": "",
  "firehose.endpoint": "",

  "flows": [
    {
      "filePattern": "/var/log/airline_tickets/*log",
      "kinesisStream": "airline_tickets",
      "partitionKeyOption": "RANDOM",
      "dataProcessingOptions": [
         {
            "optionName": "CSVTOJSON",
            "customFieldNames": ["cost","trip_class","show_to_affiliates",
                "return_date","origin","number_of_changes","gate","found_at",
                "duration","distance","destination","depart_date","actual","record_id"]
         }
      ]
    }
  ]
}

কনফিগারেশন ফাইল থেকে দেখা যায়, এজেন্ট /var/log/airline_tickets/ ডিরেক্টরিতে .log এক্সটেনশন সহ ফাইলগুলি নিরীক্ষণ করবে, সেগুলিকে পার্স করবে এবং এয়ারলাইন_টিকিট স্ট্রিমে স্থানান্তর করবে।

আমরা পরিষেবাটি পুনরায় চালু করি এবং নিশ্চিত করি যে এটি চালু এবং চলছে:

sudo service aws-kinesis-agent restart

এখন পাইথন স্ক্রিপ্ট ডাউনলোড করা যাক যা API থেকে ডেটা অনুরোধ করবে:

REPO_PATH=https://raw.githubusercontent.com/igorgorbenko/aviasales_kinesis/master/producer

wget $REPO_PATH/api_caller.py -P /home/ec2-user/
wget $REPO_PATH/requirements.txt -P /home/ec2-user/
sudo chmod a+x /home/ec2-user/api_caller.py
sudo /usr/local/bin/pip3 install -r /home/ec2-user/requirements.txt

api_caller.py স্ক্রিপ্ট Aviasales থেকে ডেটা অনুরোধ করে এবং Kinesis এজেন্ট স্ক্যান করা ডিরেক্টরিতে প্রাপ্ত প্রতিক্রিয়া সংরক্ষণ করে। এই স্ক্রিপ্টটির বাস্তবায়ন বেশ মানসম্পন্ন, একটি TicketsApi ক্লাস রয়েছে, এটি আপনাকে অ্যাসিঙ্ক্রোনাসভাবে API টানতে দেয়। আমরা একটি টোকেন সহ একটি শিরোনাম পাস করি এবং এই ক্লাসে পরামিতিগুলির অনুরোধ করি:

class TicketsApi:
    """Api caller class."""

    def __init__(self, headers):
        """Init method."""
        self.base_url = BASE_URL
        self.headers = headers

    async def get_data(self, data):
        """Get the data from API query."""
        response_json = {}
        async with ClientSession(headers=self.headers) as session:
            try:
                response = await session.get(self.base_url, data=data)
                response.raise_for_status()
                LOGGER.info('Response status %s: %s',
                            self.base_url, response.status)
                response_json = await response.json()
            except HTTPError as http_err:
                LOGGER.error('Oops! HTTP error occurred: %s', str(http_err))
            except Exception as err:
                LOGGER.error('Oops! An error ocurred: %s', str(err))
            return response_json


def prepare_request(api_token):
    """Return the headers and query fot the API request."""
    headers = {'X-Access-Token': api_token,
               'Accept-Encoding': 'gzip'}

    data = FormData()
    data.add_field('currency', CURRENCY)
    data.add_field('origin', ORIGIN)
    data.add_field('destination', DESTINATION)
    data.add_field('show_to_affiliates', SHOW_TO_AFFILIATES)
    data.add_field('trip_duration', TRIP_DURATION)
    return headers, data


async def main():
    """Get run the code."""
    if len(sys.argv) != 2:
        print('Usage: api_caller.py <your_api_token>')
        sys.exit(1)
        return
    api_token = sys.argv[1]
    headers, data = prepare_request(api_token)

    api = TicketsApi(headers)
    response = await api.get_data(data)
    if response.get('success', None):
        LOGGER.info('API has returned %s items', len(response['data']))
        try:
            count_rows = log_maker(response)
            LOGGER.info('%s rows have been saved into %s',
                        count_rows,
                        TARGET_FILE)
        except Exception as e:
            LOGGER.error('Oops! Request result was not saved to file. %s',
                         str(e))
    else:
        LOGGER.error('Oops! API request was unsuccessful %s!', response)

এজেন্টের সঠিক সেটিংস এবং কার্যকারিতা পরীক্ষা করতে, আসুন api_caller.py স্ক্রিপ্টটি পরীক্ষা করি:

sudo ./api_caller.py TOKEN

এবং আমরা এজেন্ট লগ এবং এয়ারলাইন_টিকিট ডেটা স্ট্রীমের মনিটরিং ট্যাবে কাজের ফলাফল দেখি:

tail -f /var/log/aws-kinesis-agent/aws-kinesis-agent.log

আপনি দেখতে পাচ্ছেন, সবকিছু কাজ করে এবং কাইনেসিস এজেন্ট সফলভাবে স্ট্রীমে ডেটা পাঠায়। এখন ভোক্তা কনফিগার করা যাক.

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

চলুন পুরো সিস্টেমের কেন্দ্রীয় উপাদানে যাওয়া যাক - Kinesis_analytics_airlines_app নামে Kinesis ডেটা অ্যানালিটিকসে একটি নতুন অ্যাপ্লিকেশন তৈরি করুন:

Kinesis ডেটা অ্যানালিটিক্স আপনাকে SQL ভাষা ব্যবহার করে Kinesis স্ট্রীম থেকে রিয়েল-টাইম ডেটা বিশ্লেষণ করতে দেয়। এটি একটি সম্পূর্ণ অটোস্কেলিং পরিষেবা (কাইনেসিস স্ট্রীমগুলির বিপরীতে) যা:

1. তথ্য উৎসের অনুরোধের ভিত্তিতে আপনাকে নতুন স্ট্রিম (আউটপুট স্ট্রিম) তৈরি করতে দেয়;
2. অ্যাপ্লিকেশন চালানোর সময় ঘটে যাওয়া ত্রুটি সহ একটি স্ট্রিম প্রদান করে (ত্রুটি স্ট্রিম);
3. স্বয়ংক্রিয়ভাবে ইনপুট ডেটা স্কিম নির্ধারণ করতে পারে (প্রয়োজনে এটি ম্যানুয়ালি পুনরায় সংজ্ঞায়িত করা যেতে পারে)।

এটি একটি সস্তা পরিষেবা নয় - কাজের প্রতি ঘন্টা 0.11 USD, তাই আপনার এটি সাবধানে ব্যবহার করা উচিত এবং আপনার কাজ শেষ হয়ে গেলে এটি মুছে ফেলা উচিত।

আসুন অ্যাপ্লিকেশনটিকে ডেটা উত্সের সাথে সংযুক্ত করি:

আমরা যে স্ট্রিমটির সাথে সংযোগ করতে যাচ্ছি তা নির্বাচন করুন (এয়ারলাইন_টিকিট):

এর পরে, আপনাকে একটি নতুন IAM ভূমিকা সংযুক্ত করতে হবে যাতে অ্যাপ্লিকেশনটি স্ট্রীম থেকে পড়তে এবং স্ট্রীমে লিখতে পারে। এটি করার জন্য, অ্যাক্সেস অনুমতি ব্লকে কিছু পরিবর্তন না করাই যথেষ্ট:

এখন স্ট্রিমে ডেটা স্কিমা আবিষ্কারের অনুরোধ করা যাক; এটি করতে, "ডিসকভার স্কিমা" বোতামে ক্লিক করুন। ফলস্বরূপ, আইএএম ভূমিকা আপডেট করা হবে (একটি নতুন তৈরি করা হবে) এবং ইতিমধ্যে স্ট্রীমে আসা ডেটা থেকে স্কিমা সনাক্তকরণ চালু করা হবে:

এখন আপনাকে এসকিউএল এডিটরে যেতে হবে। আপনি যখন এই বোতামটিতে ক্লিক করবেন, একটি উইন্ডো প্রদর্শিত হবে যা আপনাকে অ্যাপ্লিকেশনটি চালু করতে বলবে - আপনি কী লঞ্চ করতে চান তা নির্বাচন করুন:

এসকিউএল এডিটর উইন্ডোতে নিচের সহজ ক্যোয়ারীটি ঢোকান এবং সেভ এন্ড রান এসকিউএল এ ক্লিক করুন:

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ("cost" DOUBLE, "gate" VARCHAR(16));

CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM "cost", "gate"
FROM "SOURCE_SQL_STREAM_001"
WHERE "cost" < 5000
    and "gate" = 'Aeroflot';

রিলেশনাল ডাটাবেসে, আপনি রেকর্ড যোগ করতে INSERT স্টেটমেন্ট ব্যবহার করে টেবিলের সাথে কাজ করেন এবং ডেটা ক্যোয়ারী করার জন্য একটি SELECT স্টেটমেন্ট ব্যবহার করেন। Amazon Kinesis ডেটা অ্যানালিটিক্সে, আপনি স্ট্রীম (স্ট্রিম) এবং পাম্প (PUMPs)-এর সাথে কাজ করেন—একটি স্ট্রীম থেকে অন্য স্ট্রীমে একটি অ্যাপ্লিকেশনে ডেটা ঢোকানোর ক্রমাগত সন্নিবেশের অনুরোধ।

উপরে উপস্থাপিত এসকিউএল ক্যোয়ারী পাঁচ হাজার রুবেলের নিচে এরোফ্লট টিকিটের জন্য অনুসন্ধান করে। এই শর্তগুলি পূরণ করে এমন সমস্ত রেকর্ড DESTINATION_SQL_STREAM স্ট্রীমে স্থাপন করা হবে৷

গন্তব্য ব্লকে, বিশেষ_স্ট্রিম স্ট্রীম নির্বাচন করুন এবং ইন-অ্যাপ্লিকেশন স্ট্রীম নাম DESTINATION_SQL_STREAM ড্রপ-ডাউন তালিকায়:

সমস্ত ম্যানিপুলেশনের ফলাফল নীচের ছবির মত কিছু হওয়া উচিত:

একটি SNS বিষয় তৈরি করা এবং সদস্যতা নেওয়া

সিম্পল নোটিফিকেশন সার্ভিসে যান এবং সেখানে এয়ারলাইন্স নামে একটি নতুন বিষয় তৈরি করুন:

এই বিষয়ে সাবস্ক্রাইব করুন এবং যে মোবাইল ফোন নম্বরে SMS বিজ্ঞপ্তি পাঠানো হবে তা নির্দেশ করুন:

ডায়নামোডিবিতে একটি টেবিল তৈরি করুন

তাদের এয়ারলাইন_টিকিট স্ট্রীম থেকে কাঁচা ডেটা সঞ্চয় করতে, আসুন একই নামের সাথে ডায়নামোডিবিতে একটি টেবিল তৈরি করি। আমরা প্রাথমিক কী হিসাবে record_id ব্যবহার করব:

একটি ল্যাম্বডা ফাংশন সংগ্রাহক তৈরি করা হচ্ছে

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

সংগ্রাহক ল্যাম্বডা ফাংশনের জন্য একটি IAM ভূমিকা তৈরি করা
প্রথমে, আসুন ল্যাম্বডা-টিকেট প্রসেসিং রোল নামে ল্যাম্বডার জন্য একটি নতুন আইএএম ভূমিকা তৈরি করি:

পরীক্ষার উদাহরণের জন্য, পূর্ব-কনফিগার করা AmazonKinesisReadOnlyAccess এবং AmazonDynamoDBFullAccess নীতিগুলি বেশ উপযুক্ত, যেমনটি নীচের ছবিতে দেখানো হয়েছে:

যখন নতুন এন্ট্রি এয়ারলাইন_স্ট্রিমে প্রবেশ করে তখন এই ল্যাম্বডা কাইনেসিস থেকে একটি ট্রিগার দ্বারা চালু করা উচিত, তাই আমাদের একটি নতুন ট্রিগার যোগ করতে হবে:

যা অবশিষ্ট থাকে তা হল কোডটি সন্নিবেশ করা এবং ল্যাম্বডা সংরক্ষণ করা।

"""Parsing the stream and inserting into the DynamoDB table."""
import base64
import json
import boto3
from decimal import Decimal

DYNAMO_DB = boto3.resource('dynamodb')
TABLE_NAME = 'airline_tickets'

class TicketsParser:
    """Parsing info from the Stream."""

    def __init__(self, table_name, records):
        """Init method."""
        self.table = DYNAMO_DB.Table(table_name)
        self.json_data = TicketsParser.get_json_data(records)

    @staticmethod
    def get_json_data(records):
        """Return deserialized data from the stream."""
        decoded_record_data = ([base64.b64decode(record['kinesis']['data'])
                                for record in records])
        json_data = ([json.loads(decoded_record)
                      for decoded_record in decoded_record_data])
        return json_data

    @staticmethod
    def get_item_from_json(json_item):
        """Pre-process the json data."""
        new_item = {
            'record_id': json_item.get('record_id'),
            'cost': Decimal(json_item.get('cost')),
            'trip_class': json_item.get('trip_class'),
            'show_to_affiliates': json_item.get('show_to_affiliates'),
            'origin': json_item.get('origin'),
            'number_of_changes': int(json_item.get('number_of_changes')),
            'gate': json_item.get('gate'),
            'found_at': json_item.get('found_at'),
            'duration': int(json_item.get('duration')),
            'distance': int(json_item.get('distance')),
            'destination': json_item.get('destination'),
            'depart_date': json_item.get('depart_date'),
            'actual': json_item.get('actual')
        }
        return new_item

    def run(self):
        """Batch insert into the table."""
        with self.table.batch_writer() as batch_writer:
            for item in self.json_data:
                dynamodb_item = TicketsParser.get_item_from_json(item)
                batch_writer.put_item(dynamodb_item)

        print('Has been added ', len(self.json_data), 'items')

def lambda_handler(event, context):
    """Parse the stream and insert into the DynamoDB table."""
    print('Got event:', event)
    parser = TicketsParser(TABLE_NAME, event['Records'])
    parser.run()

একটি ল্যাম্বডা ফাংশন নোটিফায়ার তৈরি করা হচ্ছে

দ্বিতীয় ল্যাম্বডা ফাংশন, যা দ্বিতীয় স্ট্রিম (বিশেষ_স্ট্রিম) নিরীক্ষণ করবে এবং এসএনএস-এ একটি বিজ্ঞপ্তি পাঠাবে, একইভাবে তৈরি করা হয়েছে। অতএব, এই ল্যাম্বডাকে অবশ্যই কাইনেসিস থেকে পড়ার অ্যাক্সেস থাকতে হবে এবং একটি প্রদত্ত SNS বিষয়ে বার্তা পাঠাতে হবে, যা SNS পরিষেবা দ্বারা এই বিষয়ের সমস্ত গ্রাহকদের (ইমেল, এসএমএস, ইত্যাদি) পাঠানো হবে।

একটি আইএএম ভূমিকা তৈরি করা
প্রথমত, আমরা এই ল্যাম্বডার জন্য IAM ভূমিকা Lambda-KinesisAlarm তৈরি করি, এবং তারপরে তৈরি করা অ্যালার্ম_নোটিফায়ার ল্যাম্বডাকে এই ভূমিকা অর্পণ করি:

স্পেশাল_স্ট্রিমে প্রবেশের জন্য নতুন রেকর্ডের জন্য এই ল্যাম্বডা একটি ট্রিগারে কাজ করা উচিত, তাই আপনাকে ট্রিগারটিকে একইভাবে কনফিগার করতে হবে যেমনটি আমরা কালেক্টর ল্যাম্বডার জন্য করেছি।

এই ল্যাম্বডা কনফিগার করা সহজ করার জন্য, আসুন একটি নতুন পরিবেশ পরিবর্তনশীল - TOPIC_ARN, যেখানে আমরা এয়ারলাইন্স বিষয়ের ANR (Amazon Recourse Names) রাখি:

এবং ল্যাম্বডা কোড সন্নিবেশ করান, এটি মোটেও জটিল নয়:

import boto3
import base64
import os

SNS_CLIENT = boto3.client('sns')
TOPIC_ARN = os.environ['TOPIC_ARN']


def lambda_handler(event, context):
    try:
        SNS_CLIENT.publish(TopicArn=TOPIC_ARN,
                           Message='Hi! I have found an interesting stuff!',
                           Subject='Airline tickets alarm')
        print('Alarm message has been successfully delivered')
    except Exception as err:
        print('Delivery failure', str(err))

মনে হচ্ছে এখানেই ম্যানুয়াল সিস্টেম কনফিগারেশন সম্পন্ন হয়েছে। যা অবশিষ্ট থাকে তা হল পরীক্ষা করা এবং নিশ্চিত করা যে আমরা সবকিছু সঠিকভাবে কনফিগার করেছি।

Terraform কোড থেকে স্থাপন

প্রয়োজনীয় প্রস্তুতি

Terraform কোড থেকে অবকাঠামো স্থাপনের জন্য একটি খুব সুবিধাজনক ওপেন সোর্স টুল। এটির নিজস্ব সিনট্যাক্স রয়েছে যা শিখতে সহজ এবং কীভাবে এবং কী স্থাপন করতে হবে তার অনেক উদাহরণ রয়েছে। এটম এডিটর বা ভিজ্যুয়াল স্টুডিও কোডে অনেক সুবিধাজনক প্লাগইন রয়েছে যা টেরাফর্মের সাথে কাজ করা সহজ করে তোলে।

আপনি বিতরণ ডাউনলোড করতে পারেন এখানে থেকে. সমস্ত Terraform ক্ষমতাগুলির একটি বিশদ বিশ্লেষণ এই নিবন্ধের সুযোগের বাইরে, তাই আমরা নিজেদেরকে মূল পয়েন্টগুলিতে সীমাবদ্ধ করব।

কীভাবে চালাবেন

প্রকল্পের সম্পূর্ণ কোড হল আমার ভান্ডারে. আমরা নিজেদের কাছে সংগ্রহস্থল ক্লোন. শুরু করার আগে, আপনাকে নিশ্চিত করতে হবে যে আপনার AWS CLI ইনস্টল এবং কনফিগার করা আছে, কারণ... Terraform ~/.aws/credentials ফাইলে শংসাপত্রের সন্ধান করবে।

টেরাফর্ম বর্তমানে ক্লাউডে আমাদের জন্য কী তৈরি করছে তা দেখার জন্য সম্পূর্ণ পরিকাঠামো স্থাপনের আগে পরিকল্পনা কমান্ড চালানো একটি ভাল অনুশীলন:

terraform.exe plan

বিজ্ঞপ্তি পাঠানোর জন্য আপনাকে একটি ফোন নম্বর লিখতে বলা হবে। এই পর্যায়ে এটি প্রবেশ করা আবশ্যক নয়।

প্রোগ্রামের অপারেশন পরিকল্পনা বিশ্লেষণ করে, আমরা সংস্থান তৈরি করা শুরু করতে পারি:

terraform.exe apply

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

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

অ্যাপ্লিকেশন চালু করুন:

এর পরে, ড্রপ-ডাউন তালিকা থেকে নির্বাচন করে আপনাকে স্পষ্টভাবে ইন-অ্যাপ্লিকেশন স্ট্রিম নাম সেট করতে হবে:

এখন সবকিছু যেতে প্রস্তুত.

আবেদন পরীক্ষা করা হচ্ছে

আপনি ম্যানুয়ালি বা Terraform কোডের মাধ্যমে সিস্টেমটি যেভাবে স্থাপন করেছেন তা নির্বিশেষে, এটি একই কাজ করবে।

আমরা SSH-এর মাধ্যমে EC2 ভার্চুয়াল মেশিনে লগ ইন করি যেখানে Kinesis Agent ইনস্টল করা আছে এবং api_caller.py স্ক্রিপ্ট চালান

sudo ./api_caller.py TOKEN

আপনাকে যা করতে হবে তা হল আপনার নম্বরে একটি SMS এর জন্য অপেক্ষা করুন:

SMS - প্রায় 1 মিনিটের মধ্যে ফোনে বার্তাটি আসে:

পরবর্তী, আরও বিশদ বিশ্লেষণের জন্য রেকর্ডগুলি DynamoDB ডাটাবেসে সংরক্ষিত হয়েছে কিনা তা দেখতে বাকি রয়েছে। এয়ারলাইন_টিকিট টেবিলে প্রায় নিম্নলিখিত ডেটা রয়েছে:

উপসংহার

কাজ চলাকালীন, অ্যামাজন কাইনেসিসের উপর ভিত্তি করে একটি অনলাইন ডেটা প্রসেসিং সিস্টেম তৈরি করা হয়েছিল। এসকিউএল কমান্ড ব্যবহার করে কাইনেসিস ডেটা স্ট্রীম এবং রিয়েল-টাইম অ্যানালিটিক্স কাইনেসিস অ্যানালিটিক্সের সাথে কাইনেসিস এজেন্ট ব্যবহার করার বিকল্পগুলি, সেইসাথে অন্যান্য AWS পরিষেবাগুলির সাথে অ্যামাজন কাইনেসিসের মিথস্ক্রিয়া বিবেচনা করা হয়েছিল।

আমরা উপরের সিস্টেমটিকে দুটি উপায়ে স্থাপন করেছি: একটি বরং দীর্ঘ ম্যানুয়াল এবং একটি দ্রুত টেরাফর্ম কোড থেকে।

সমস্ত প্রকল্প উত্স কোড উপলব্ধ আমার GitHub সংগ্রহস্থলে, আমি আপনাকে এটির সাথে নিজেকে পরিচিত করার পরামর্শ দিচ্ছি।

আমি নিবন্ধটি নিয়ে আলোচনা করতে পেরে খুশি, আমি আপনার মন্তব্যের জন্য অপেক্ষা করছি। গঠনমূলক সমালোচনা আশা করছি।

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

উত্স: www.habr.com