Amazon Kinesis और सर्वर रहित सादगी के साथ Aviasales एपीआई एकीकरण

परिचय

इस लेख का मुख्य उद्देश्य एडब्ल्यूएस में स्ट्रीमिंग जानकारी के उपयोग की एक सामान्य समझ देना है, हम इसे इस सवाल से बाहर कर देते हैं कि एपीआई द्वारा उपयोग किया गया डेटा कड़ाई से प्रासंगिक नहीं है और कैश से स्थानांतरित किया जाता है, जो कि Aviasales.ru और Jetradar.com साइटों के उपयोगकर्ताओं की खोजों के आधार पर उत्पन्न होता है। पिछले 48 घंटे।

• मैनुअल - एडब्ल्यूएस प्रबंधन कंसोल के माध्यम से;
• टेराफॉर्म कोड से आधारभूत संरचना - आलसी स्वचालन इंजीनियरों के लिए;

विकास के तहत प्रणाली का आर्किटेक्चर

• Aviasales API - इस API द्वारा दिए गए डेटा का उपयोग बाद के सभी कामों के लिए किया जाएगा;
• EC2 निर्माता उदाहरण - क्लाउड में एक नियमित वर्चुअल मशीन जिस पर इनपुट डेटा स्ट्रीम उत्पन्न होगी:
  
  
  • Kinesis Agent एक स्थानीय रूप से स्थापित जावा एप्लिकेशन है जो Kinesis (Kinesis Data Streams या Kinesis Firehose) को डेटा एकत्र करने और भेजने का एक सरल तरीका प्रदान करता है। एजेंट लगातार निर्दिष्ट निर्देशिकाओं में फ़ाइलों के एक सेट की निगरानी करता है और किनिस को नया डेटा भेजता है;
  • कॉलर एपीआई स्क्रिप्ट - एक पायथन स्क्रिप्ट जो एपीआई अनुरोध करती है और एक फ़ोल्डर में प्रतिक्रिया डालती है जो किनेसिस एजेंट मॉनिटर करता है;
• Kinesis Data Streams — ;
• Kinesis Analytics — , . Amazon Kinesis Data Analytics ;
• AWS Lambda — , . ;
• Amazon DynamoDB — «‑» , 10 . DynamoDB - , . DynamoDB , . ;
• Amazon SNS — « — » (Pub/Sub), , . SNS push-, SMS- .

प्रारंभिक प्रशिक्षण

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

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

यह माना जाता है कि आपके पास पहले से ही AWS खाता है। मैं इस बात पर विशेष ध्यान देना चाहता हूं कि Kinesis और SMS के माध्यम से सूचनाएं भेजना वार्षिक Free Tier (मुफ्त उपयोग) में शामिल नहीं है । लेकिन इसके बावजूद, कुछ डॉलर को ध्यान में रखते हुए, प्रस्तावित प्रणाली का निर्माण और इसके साथ खेलना काफी संभव है। और, ज़ाहिर है, अनावश्यक होने के बाद सभी संसाधनों को हटाने के लिए मत भूलना।

मैनुअल तैनाती प्रणाली

Kinesis डेटा स्ट्रीम सेट करना

निर्माता सेटिंग

EC2 के लिए IAM रोल्स बनाना

, , EC2 Permissions:

, : AmazonKinesisFullAccess CloudWatchFullAccess.

- , : EC2-KinesisStreams-FullAccess. , , :

, :

.

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

sudo mkdir /var/log/airline_tickets

sudo vim /etc/aws-kinesis/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"]
         }
      ]
    }
  ]
}

sudo service aws-kinesis-agent restart

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

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)

sudo ./api_caller.py TOKEN

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

Kinesis Data Analytics की स्थापना

1. आपको स्रोत डेटा के प्रश्नों पर आधारित नई स्ट्रीम (आउटपुट स्ट्रीम) बनाने की अनुमति देता है;
2. एप्लिकेशन ऑपरेशन (त्रुटि स्ट्रीम) के दौरान हुई त्रुटियों के साथ एक स्ट्रीम प्रदान करता है;
3. यह स्वचालित रूप से इनपुट डेटा योजना निर्धारित कर सकता है (यदि आवश्यक हो तो इसे मैन्युअल रूप से फिर से परिभाषित किया जा सकता है)।

यह एक महंगी सेवा है - प्रति घंटे 0.11 USD, इसलिए आपको इसे सावधानी से उपयोग करना चाहिए और काम पूरा होने पर इसे हटा देना चाहिए।

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 स्टेटमेंट का उपयोग करते हुए तालिकाओं के साथ काम करते हैं और क्वेरी डेटा का चयन करें। Amazon Kinesis Data Analytics में, आप स्ट्रीम (STREAM) और "पंप" (PUMP) के साथ काम करते हैं - निरंतर सम्मिलन अनुरोध जो किसी एप्लिकेशन में डेटा को एक स्ट्रीम से दूसरी स्ट्रीम में सम्मिलित करते हैं।

SNS विषय को बनाना और सदस्यता लेना

डायनेमोडीबी में एक तालिका बनाना

एक लंबोदर कलेक्टर फ़ंक्शन बनाना

एक लंबो कलेक्टर फ़ंक्शन के लिए IAM भूमिका बनाना

IAM Lambda-TicketsProcessingRole:

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

एक लंबोदर फ़ंक्शन नोटिफ़ायर बनाना

IAM रोल्स बनाएं

IAM Lambda-KinesisAlarm , alarm_notifier:

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.exe plan

terraform.exe apply

आवेदन परीक्षण

sudo ./api_caller.py TOKEN

निष्कर्ष