Aviasales API ინტეგრაცია Amazon Kinesis-თან და სერვერის სიმარტივით

ჰეი ჰაბრ!

მოგწონთ თვითმფრინავების ფრენა? მე მიყვარს, მაგრამ თვითიზოლაციის დროს მეც შემიყვარდა ავიაბილეთების მონაცემების ანალიზი ერთი ცნობილი რესურსიდან - Aviasales.

დღეს ჩვენ გავაანალიზებთ Amazon Kinesis-ის მუშაობას, ავაშენებთ სტრიმინგ სისტემას რეალურ დროში ანალიტიკით, დავაინსტალირებთ Amazon DynamoDB NoSQL მონაცემთა ბაზას, როგორც მონაცემთა ძირითად შესანახად და დავაყენებთ SMS შეტყობინებებს საინტერესო ბილეთებისთვის.

ყველა დეტალი ჭრის ქვეშაა! წადი!

შესავალი

მაგალითად, ჩვენ გვჭირდება წვდომა Aviasales API. მასზე წვდომა უფასოა და შეზღუდვების გარეშე; თქვენ უბრალოდ უნდა დარეგისტრირდეთ "დეველოპერების" განყოფილებაში, რათა მიიღოთ თქვენი API ჟეტონი, რომ მიიღოთ მონაცემები.

ამ სტატიის მთავარი მიზანია AWS-ში ინფორმაციის ნაკადის გამოყენების ზოგადი გაგება; ჩვენ გავითვალისწინებთ, რომ გამოყენებული API-ს მიერ დაბრუნებული მონაცემები არ არის მკაცრად განახლებული და გადაცემულია ქეშიდან, რომელიც არის ჩამოყალიბდა Aviasales.ru-სა და Jetradar.com-ის მომხმარებლების მიერ ბოლო 48 საათის განმავლობაში ძიების საფუძველზე.

Kinesis-აგენტი, რომელიც დაინსტალირებულია მწარმოებელ მანქანაზე, მიღებული API-ით ავტომატურად აანალიზებს და გადასცემს მონაცემებს სასურველ ნაკადზე Kinesis Data Analytics-ის საშუალებით. ამ ნაკადის დაუმუშავებელი ვერსია დაიწერება პირდაპირ მაღაზიაში. DynamoDB-ში განლაგებული ნედლეული მონაცემთა შენახვა საშუალებას მისცემს ბილეთების უფრო ღრმა ანალიზს BI ინსტრუმენტების მეშვეობით, როგორიცაა AWS Quick Sight.

ჩვენ განვიხილავთ ორ ვარიანტს მთელი ინფრასტრუქტურის განლაგებისთვის:

• სახელმძღვანელო - AWS Management Console-ის მეშვეობით;
• Terraform კოდის ინფრასტრუქტურა განკუთვნილია ზარმაცი ავტომატებისთვის;

განვითარებული სისტემის არქიტექტურა

გამოყენებული კომპონენტები:

• Aviasales API — ამ API-ს მიერ დაბრუნებული მონაცემები გამოყენებული იქნება ყველა შემდგომი სამუშაოსთვის;
• EC2 პროდიუსერის მაგალითი — ჩვეულებრივი ვირტუალური მანქანა ღრუბელში, რომელზედაც წარმოიქმნება შეყვანის მონაცემთა ნაკადი:
  • Kinesis აგენტი არის მანქანაზე ადგილობრივად დაინსტალირებული Java აპლიკაცია, რომელიც უზრუნველყოფს მონაცემთა შეგროვებისა და გაგზავნის მარტივ გზას Kinesis-ში (Kinesis Data Streams ან Kinesis Firehose). აგენტი მუდმივად აკონტროლებს ფაილების ერთობლიობას მითითებულ დირექტორიაში და აგზავნის ახალ მონაცემებს Kinesis-ში;
  • API Caller Script - პითონის სკრიპტი, რომელიც მიმართავს API-ს მოთხოვნებს და პასუხს ათავსებს საქაღალდეში, რომელსაც აკონტროლებს Kinesis Agent;
• Kinesis მონაცემთა ნაკადები — რეალურ დროში მონაცემთა ნაკადის სერვისი ფართო მასშტაბის შესაძლებლობებით;
• Kinesis Analytics არის უსერვერო სერვისი, რომელიც ამარტივებს სტრიმინგის მონაცემების ანალიზს რეალურ დროში. Amazon Kinesis Data Analytics აკონფიგურირებს აპლიკაციის რესურსებს და ავტომატურად მასშტაბირებს შემომავალი მონაცემების ნებისმიერი მოცულობის დასამუშავებლად;
• AWS ლამბდა — სერვისი, რომელიც საშუალებას გაძლევთ გაუშვათ კოდი სერვერების სარეზერვო ასლის ან დაყენების გარეშე. ყველა გამოთვლითი სიმძლავრე ავტომატურად მასშტაბირებულია თითოეული ზარისთვის;
• Amazon DynamoDB - გასაღები-მნიშვნელობის წყვილებისა და დოკუმენტების მონაცემთა ბაზა, რომელიც უზრუნველყოფს 10 მილიწამზე ნაკლებ შეყოვნებას ნებისმიერი მასშტაბის გაშვებისას. DynamoDB-ის გამოყენებისას, თქვენ არ გჭირდებათ რაიმე სერვერის უზრუნველყოფა, დაყენება ან მართვა. DynamoDB ავტომატურად ადიდებს ცხრილებს, რათა შეცვალოს არსებული რესურსების რაოდენობა და შეინარჩუნოს მაღალი შესრულება. არ არის საჭირო სისტემის ადმინისტრირება;
• Amazon SNS - სრულად მართული სერვისი შეტყობინებების გაგზავნისთვის გამომცემელი-აბონენტის (Pub/Sub) მოდელის გამოყენებით, რომლითაც შეგიძლიათ მიკროსერვისების, განაწილებული სისტემების და სერვერის გარეშე აპლიკაციების იზოლირება. SNS შეიძლება გამოყენებულ იქნას საბოლოო მომხმარებლებისთვის ინფორმაციის გასაგზავნად მობილური Push შეტყობინებების, SMS შეტყობინებების და ელ.ფოსტის საშუალებით.

საწყისი ტრენინგი

მონაცემთა ნაკადის ემულაციისთვის გადავწყვიტე გამომეყენებინა Aviasales API-ს მიერ დაბრუნებული ავიაბილეთის ინფორმაცია. IN დოკუმენტაცია სხვადასხვა მეთოდების საკმაოდ ვრცელი ჩამონათვალი, ავიღოთ ერთ-ერთი მათგანი - „თვიური ფასების კალენდარი“, რომელიც აბრუნებს ფასებს თვის თითოეული დღისთვის, დაჯგუფებული გადარიცხვების რაოდენობის მიხედვით. თუ მოთხოვნაში არ მიუთითებთ ძიების თვეს, ინფორმაცია დაბრუნდება მიმდინარე თვის მომდევნო თვისთვის.

ასე რომ, მოდით დავრეგისტრირდეთ და მივიღოთ ჩვენი ნიშანი.

მოთხოვნის მაგალითი მოცემულია ქვემოთ:

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 ანგარიში. მინდა დაუყოვნებლივ გავამახვილო განსაკუთრებული ყურადღება იმ ფაქტზე, რომ Kinesis და SMS-ით შეტყობინებების გაგზავნა არ შედის ყოველწლიურ უფასო დონე (უფასო გამოყენება). მაგრამ ამის მიუხედავად, რამდენიმე დოლარის გათვალისწინებით, სავსებით შესაძლებელია შემოთავაზებული სისტემის აშენება და მასთან თამაში. და, რა თქმა უნდა, არ დაგავიწყდეთ წაშალოთ ყველა რესურსი მას შემდეგ, რაც ისინი აღარ დაგჭირდებათ.

საბედნიეროდ, DynamoDb და ლამბდა ფუნქციები ჩვენთვის უფასო იქნება, თუ თქვენ დააკმაყოფილებთ ყოველთვიურ უფასო ლიმიტებს. მაგალითად, DynamoDB-სთვის: 25 GB საცავი, 25 WCU/RCU და 100 მილიონი მოთხოვნა. და მილიონი ლამბდა ფუნქციის ზარი თვეში.

სისტემის ხელით განლაგება

Kinesis მონაცემთა ნაკადების დაყენება

მოდით გადავიდეთ Kinesis Data Streams სერვისზე და შევქმნათ ორი ახალი ნაკადი, თითო ფრაგმენტი თითოეულისთვის.

რა არის ნატეხი?
ნატეხი არის Amazon Kinesis ნაკადის მონაცემთა გადაცემის ძირითადი ერთეული. ერთი სეგმენტი უზრუნველყოფს შეყვანის მონაცემთა გადაცემას 1 მბ/წმ სიჩქარით და გამომავალი მონაცემების გადაცემას 2 მბ/წმ სიჩქარით. ერთი სეგმენტი მხარს უჭერს 1000-მდე PUT ჩანაწერს წამში. მონაცემთა ნაკადის შექმნისას, თქვენ უნდა მიუთითოთ სეგმენტების საჭირო რაოდენობა. მაგალითად, შეგიძლიათ შექმნათ მონაცემთა ნაკადი ორი სეგმენტით. მონაცემთა ეს ნაკადი უზრუნველყოფს შეყვანის მონაცემთა გადაცემას 2 მბ/წმ სიჩქარით და გამომავალი მონაცემების გადაცემას 4 მბ/წმ, 2000-მდე PUT ჩანაწერის მხარდაჭერა წამში.

რაც უფრო მეტი ფრაგმენტია თქვენს ნაკადში, მით მეტია მისი გამტარუნარიანობა. პრინციპში, ასე ხდება ნაკადების მასშტაბირება - ნამსხვრევების დამატებით. მაგრამ რაც მეტი ნატეხი გაქვთ, მით უფრო მაღალია ფასი. თითოეული ნატეხი ღირს 1,5 ცენტი საათში და დამატებითი 1.4 ცენტი ყოველი მილიონი PUT დატვირთვის ერთეულზე.

მოდით შევქმნათ ახალი ნაკადი სახელით ავია ბილეთები1 ნაჭერი საკმარისი იქნება მისთვის:

ახლა მოდით შევქმნათ სხვა თემა სახელით სპეციალური_ნაკადი:

პროდიუსერის დაყენება

დავალების გასაანალიზებლად საკმარისია გამოიყენოთ ჩვეულებრივი EC2 ინსტანცია, როგორც მონაცემთა მწარმოებელი. ეს არ უნდა იყოს მძლავრი, ძვირადღირებული ვირტუალური მანქანა; spot t2.micro კარგად იმუშავებს.

მნიშვნელოვანი შენიშვნა: მაგალითად, თქვენ უნდა გამოიყენოთ სურათი - Amazon Linux AMI 2018.03.0, მას აქვს ნაკლები პარამეტრი Kinesis Agent-ის სწრაფად გასაშვებად.

გადადით EC2 სერვისზე, შექმენით ახალი ვირტუალური მანქანა, აირჩიეთ სასურველი AMI ტიპის t2.micro, რომელიც შედის უფასო იარუსში:

იმისათვის, რომ ახლად შექმნილ ვირტუალურ მანქანას შეეძლოს Kinesis სერვისთან ურთიერთობა, მას ამის უფლება უნდა მიენიჭოს. ამის საუკეთესო გზაა IAM როლის მინიჭება. ამიტომ, ნაბიჯი 3: ინსტანციის დეტალების კონფიგურაციის ეკრანზე, თქვენ უნდა აირჩიოთ შექმენით ახალი IAM როლი:

IAM როლის შექმნა EC2-სთვის

ფანჯარაში, რომელიც იხსნება, აირჩიეთ, რომ ჩვენ ვქმნით ახალ როლს EC2-სთვის და გადადით ნებართვების განყოფილებაში:

ტრენინგის მაგალითის გამოყენებით, ჩვენ არ მოგვიწევს რესურსების უფლებების მარცვლოვანი კონფიგურაციის ყველა სირთულეში შემოსვლა, ამიტომ ჩვენ ვირჩევთ Amazon-ის მიერ წინასწარ კონფიგურირებულ პოლიტიკას: AmazonKinesisFullAccess და CloudWatchFullAccess.

მოდით მივცეთ რაიმე მნიშვნელოვანი სახელი ამ როლს, მაგალითად: EC2-KinesisStreams-FullAccess. შედეგი უნდა იყოს იგივე, რაც ნაჩვენებია ქვემოთ მოცემულ სურათზე:

ამ ახალი როლის შექმნის შემდეგ, არ დაგავიწყდეთ მისი მიმაგრება შექმნილ ვირტუალური მანქანის მაგალითზე:

ჩვენ სხვას არაფერს ვცვლით ამ ეკრანზე და გადავდივართ შემდეგ ფანჯრებზე.

მყარი დისკის პარამეტრები შეიძლება დარჩეს ნაგულისხმევად, ისევე როგორც ტეგები (თუმცა კარგი პრაქტიკაა ტეგების გამოყენება, ყოველ შემთხვევაში, მაგალითს მიეცით სახელი და მიუთითეთ გარემო).

ახლა ჩვენ ვიმყოფებით ნაბიჯი 6: უსაფრთხოების ჯგუფის კონფიგურაციის ჩანართზე, სადაც თქვენ უნდა შექმნათ ახალი ან მიუთითოთ არსებული უსაფრთხოების ჯგუფი, რომელიც საშუალებას გაძლევთ დაუკავშირდეთ ssh (პორტი 22) ინსტანციას. აირჩიეთ წყარო -> ჩემი IP იქ და შეგიძლიათ გაუშვათ ეგზემპლარი.

როგორც კი ის გადადის გაშვებულ სტატუსზე, შეგიძლიათ სცადოთ მასთან დაკავშირება ssh-ის საშუალებით.

იმისათვის, რომ შეძლოთ Kinesis Agent-თან მუშაობა, მანქანასთან წარმატებით დაკავშირების შემდეგ, თქვენ უნდა შეიყვანოთ შემდეგი ბრძანებები ტერმინალში:

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"]
         }
      ]
    }
  ]
}

როგორც კონფიგურაციის ფაილიდან ჩანს, აგენტი მონიტორინგს გაუწევს ფაილებს .log გაფართოებით /var/log/airline_tickets/ დირექტორიაში, გააანალიზებს მათ და გადასცემს airline_tickets ნაკადში.

ჩვენ გადავტვირთავთ სერვისს და ვრწმუნდებით, რომ ის მუშაობს და მუშაობს:

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 Agent წარმატებით აგზავნის მონაცემებს ნაკადში. ახლა მოდით დავაკონფიგურიროთ მომხმარებელი.

Kinesis Data Analytics-ის დაყენება

მოდით გადავიდეთ მთელი სისტემის ცენტრალურ კომპონენტზე - შექმენით ახალი აპლიკაცია Kinesis Data Analytics-ში, სახელად kinesis_analytics_airlines_app:

Kinesis Data Analytics საშუალებას გაძლევთ განახორციელოთ მონაცემთა რეალურ დროში ანალიტიკა Kinesis Streams-დან SQL ენის გამოყენებით. ეს არის სრულად ავტოსკალირების სერვისი (განსხვავებით Kinesis Streams-ისგან), რომელიც:

1. საშუალებას გაძლევთ შექმნათ ახალი ნაკადები (Output Stream) წყაროს მონაცემების მოთხოვნის საფუძველზე;
2. უზრუნველყოფს ნაკადს შეცდომებით, რომლებიც წარმოიშვა აპლიკაციების მუშაობის დროს (Error Stream);
3. შეუძლია ავტომატურად განსაზღვროს შეყვანის მონაცემთა სქემა (საჭიროების შემთხვევაში შესაძლებელია ხელით ხელახლა განსაზღვრა).

ეს არ არის იაფი სერვისი - 0.11 აშშ დოლარი სამუშაო საათში, ასე რომ თქვენ უნდა გამოიყენოთ იგი ფრთხილად და წაშალოთ, როდესაც დაასრულებთ.

მოდით დავუკავშიროთ აპლიკაცია მონაცემთა წყაროს:

აირჩიეთ ნაკადი, რომელსაც ვაპირებთ დაკავშირებას (ავიაკომპანია_ბილეთები):

შემდეგი, თქვენ უნდა დაურთოთ ახალი IAM როლი, რათა აპლიკაციამ შეძლოს წაიკითხოს ნაკადიდან და ჩაწეროს ნაკადში. ამისათვის საკმარისია არაფერი შეცვალოთ წვდომის ნებართვების ბლოკში:

ახლა მოვითხოვოთ მონაცემთა სქემის აღმოჩენა ნაკადში; ამისათვის დააწკაპუნეთ ღილაკზე „Discover schema“. შედეგად, IAM როლი განახლდება (შეიქმნება ახალი) და სქემის ამოცნობა დაიწყება ნაკადში უკვე შემოსული მონაცემებიდან:

ახლა თქვენ უნდა გადახვიდეთ SQL რედაქტორში. ამ ღილაკზე დაწკაპუნებისას გამოჩნდება ფანჯარა, რომელიც გთხოვთ აპლიკაციის გაშვებას - აირჩიეთ რის გაშვება გსურთ:

ჩადეთ შემდეგი მარტივი მოთხოვნა SQL რედაქტორის ფანჯარაში და დააჭირეთ Save and Run SQL:

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 Data Analytics-ში თქვენ მუშაობთ ნაკადებთან (STREAM) და ტუმბოებთან (PUMPs) - უწყვეტი ჩასმის მოთხოვნები, რომლებიც აპლიკაციის ერთი ნაკადიდან მონაცემებს სხვა ნაკადში ათავსებენ.

ზემოთ წარმოდგენილი SQL მოთხოვნა ეძებს აეროფლოტის ბილეთებს ხუთი ათას რუბლზე დაბალი ღირებულებით. ყველა ჩანაწერი, რომელიც აკმაყოფილებს ამ პირობებს, განთავსდება DESTINATION_SQL_STREAM ნაკადში.

დანიშნულების ბლოკში აირჩიეთ special_stream ნაკადი, ხოლო აპლიკაციის ნაკადის სახელი DESTINATION_SQL_STREAM ჩამოსაშლელ სიაში:

ყველა მანიპულაციის შედეგი უნდა იყოს მსგავსი სურათის მსგავსი:

SNS თემის შექმნა და გამოწერა

გადადით მარტივი შეტყობინებების სერვისში და შექმენით ახალი თემა, სახელწოდებით Airlines:

გამოიწერეთ ეს თემა და მიუთითეთ მობილური ტელეფონის ნომერი, რომელზეც გაიგზავნება SMS შეტყობინებები:

შექმენით ცხრილი DynamoDB-ში

მათი airline_tickets ნაკადიდან ნედლეული მონაცემების შესანახად, შევქმნათ ცხრილი DynamoDB-ში იმავე სახელით. ჩვენ გამოვიყენებთ record_id როგორც პირველადი გასაღები:

ლამბდა ფუნქციის კოლექტორის შექმნა

მოდით შევქმნათ ლამბდა ფუნქცია სახელწოდებით კოლექტორი, რომლის ამოცანა იქნება ავიაკომპანიის_ბილეთების ნაკადის გამოკითხვა და თუ იქ ახალი ჩანაწერები აღმოჩნდება, ჩასვით ეს ჩანაწერები DynamoDB ცხრილში. ცხადია, ნაგულისხმევი უფლებების გარდა, ამ ლამბდას უნდა ჰქონდეს წაკითხვის წვდომა Kinesis მონაცემთა ნაკადზე და ჩაწერის წვდომა DynamoDB-ზე.

IAM როლის შექმნა კოლექტორის ლამბდა ფუნქციისთვის
პირველი, მოდით შევქმნათ ახალი IAM როლი ლამბდასთვის, სახელად Lambda-TicketsProcessingRole:

ტესტის მაგალითისთვის, წინასწარ კონფიგურირებული AmazonKinesisReadOnlyAccess და AmazonDynamoDBFullAccess პოლიტიკა საკმაოდ შესაფერისია, როგორც ეს ნაჩვენებია ქვემოთ მოცემულ სურათზე:

ეს ლამბდა უნდა გაშვებული იყოს Kinesis-ის ტრიგერით, როდესაც ახალი ჩანაწერები შედის airline_stream-ში, ამიტომ ჩვენ უნდა დავამატოთ ახალი ტრიგერი:

რჩება მხოლოდ კოდის ჩასმა და ლამბდას შენახვა.

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

ლამბდა ფუნქციის შეტყობინებების შექმნა

ანალოგიურად იქმნება მეორე ლამბდა ფუნქცია, რომელიც მონიტორინგს გაუწევს მეორე ნაკადს (special_stream) და გაგზავნის შეტყობინებას SNS-ში. მაშასადამე, ამ ლამბდას უნდა ჰქონდეს წვდომა Kinesis-დან წასაკითხად და მოცემულ SNS თემაში შეტყობინებების გასაგზავნად, რომლებიც შემდეგ SNS სერვისის მიერ გაეგზავნება ამ თემის ყველა აბონენტს (ელფოსტა, SMS და ა.შ.).

IAM როლის შექმნა
ჯერ ვქმნით IAM როლს Lambda-KinesisAlarm ამ ლამბდასთვის და შემდეგ ამ როლს ვანიჭებთ alarm_notifier lambda-ს, რომელიც იქმნება:

ამ ლამბდამ უნდა იმუშაოს ტრიგერზე, რომ ახალი ჩანაწერები შევიდეს special_stream-ში, ასე რომ თქვენ უნდა დააკონფიგურიროთ ტრიგერი ისევე, როგორც ჩვენ გავაკეთეთ კოლექტორის ლამბდასთვის.

ამ ლამბდას კონფიგურაციის გასაადვილებლად, მოდით შემოვიტანოთ გარემოს ახალი ცვლადი - 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 არის ძალიან მოსახერხებელი ღია კოდის ინსტრუმენტი კოდიდან ინფრასტრუქტურის განსათავსებლად. მას აქვს საკუთარი სინტაქსი, რომლის სწავლაც მარტივია და აქვს მრავალი მაგალითი იმისა, თუ როგორ და რა უნდა განათავსოს. Atom რედაქტორს ან Visual Studio Code-ს აქვს მრავალი მოსახერხებელი დანამატი, რომელიც აადვილებს Terraform-თან მუშაობას.

შეგიძლიათ ჩამოტვირთოთ დისტრიბუცია აქედან გამომდინარე,. Terraform-ის ყველა შესაძლებლობის დეტალური ანალიზი სცილდება ამ სტატიის ფარგლებს, ამიტომ ჩვენ შემოვიფარგლებით ძირითადი პუნქტებით.

როგორ გარბოდა

პროექტის სრული კოდია ჩემს საცავში. ჩვენ ვაკლებთ საცავს საკუთარ თავს. დაწყებამდე უნდა დარწმუნდეთ, რომ დაინსტალირებული და კონფიგურირებული გაქვთ AWS CLI, რადგან... Terraform მოძებნის სერთიფიკატებს ~/.aws/credentials ფაილში.

კარგი პრაქტიკაა გეგმის ბრძანების გაშვება მთელი ინფრასტრუქტურის განლაგებამდე, რათა ნახოთ რას ქმნის ამჟამად Terraform ჩვენთვის ღრუბელში:

terraform.exe plan

თქვენ მოგეთხოვებათ შეიყვანოთ ტელეფონის ნომერი შეტყობინებების გასაგზავნად. ამ ეტაპზე მისი შეყვანა აუცილებელი არ არის.

პროგრამის საოპერაციო გეგმის გაანალიზების შემდეგ, შეგვიძლია დავიწყოთ რესურსების შექმნა:

terraform.exe apply

ამ ბრძანების გაგზავნის შემდეგ, თქვენ კვლავ მოგეთხოვებათ შეიყვანოთ ტელეფონის ნომერი; აკრიფეთ "დიახ", როდესაც გამოჩნდება შეკითხვა მოქმედებების რეალურად შესრულების შესახებ. ეს საშუალებას მოგცემთ დააყენოთ მთელი ინფრასტრუქტურა, განახორციელოთ EC2-ის ყველა საჭირო კონფიგურაცია, განათავსოთ ლამბდა ფუნქციები და ა.შ.

მას შემდეგ, რაც ყველა რესურსი წარმატებით შეიქმნა Terraform კოდის საშუალებით, თქვენ უნდა შეხვიდეთ Kinesis Analytics აპლიკაციის დეტალებში (სამწუხაროდ, მე ვერ ვიპოვე როგორ გავაკეთო ეს პირდაპირ კოდიდან).

გაუშვით აპლიკაცია:

ამის შემდეგ, თქვენ უნდა დააყენოთ აპლიკაციის შიგნით ნაკადის სახელი ჩამოსაშლელი სიიდან არჩევით:

ახლა ყველაფერი მზად არის წასასვლელად.

აპლიკაციის ტესტირება

მიუხედავად იმისა, თუ როგორ განათავსეთ სისტემა, ხელით თუ Terraform კოდის საშუალებით, ის იგივე იმუშავებს.

ჩვენ შევდივართ SSH-ის საშუალებით EC2 ვირტუალურ მანქანაში, სადაც დაინსტალირებულია Kinesis Agent და ვაწარმოებთ api_caller.py სკრიპტს.

sudo ./api_caller.py TOKEN

თქვენ მხოლოდ უნდა დაელოდოთ SMS თქვენს ნომერს:

SMS - შეტყობინება ტელეფონზე მოდის თითქმის 1 წუთში:

რჩება თუ არა ჩანაწერები შენახული DynamoDB მონაცემთა ბაზაში შემდგომი, უფრო დეტალური ანალიზისთვის. ავიაკომპანიის_ბილეთების ცხრილი შეიცავს დაახლოებით შემდეგ მონაცემებს:

დასკვნა

შესრულებული სამუშაოს დროს აშენდა ონლაინ მონაცემთა დამუშავების სისტემა Amazon Kinesis-ის საფუძველზე. განხილული იყო Kinesis Agent-ის გამოყენების ვარიანტები Kinesis Data Streams-თან და რეალურ დროში ანალიტიკასთან Kinesis Analytics SQL ბრძანებების გამოყენებით, ისევე როგორც Amazon Kinesis-ის ურთიერთქმედება სხვა AWS სერვისებთან.

ჩვენ განვათავსეთ ზემოაღნიშნული სისტემა ორი გზით: საკმაოდ გრძელი სახელმძღვანელო და სწრაფი Terraform კოდიდან.

ყველა პროექტის წყაროს კოდი ხელმისაწვდომია ჩემს GitHub საცავში, გირჩევთ გაეცნოთ მას.

სიამოვნებით განვიხილავ სტატიას, ველი თქვენს კომენტარებს. კონსტრუქციული კრიტიკის იმედი მაქვს.

გისურვებთ წარმატებას!

წყარო: www.habr.com