تكامل Aviasales API مع Amazon Kinesis والبساطة بدون خادم

المقدمة

الغرض الرئيسي من هذه المقالة هو إعطاء فهم عام لاستخدام معلومات التدفق في AWS ، ونجعل من غير المستبعد أن البيانات التي يتم إرجاعها بواسطة واجهة برمجة التطبيقات المستخدمة ليست ذات صلة صارمة ويتم نقلها من ذاكرة التخزين المؤقت ، والتي يتم إنشاؤها استنادًا إلى عمليات البحث عن مستخدمي مواقع Aviasales.ru و Jetradar.com آخر 48 ساعة.

• دليل - من خلال وحدة تحكم إدارة AWS ؛
• البنية التحتية من كود Terraform - لمهندسي الأتمتة البطيئة ؛

بنية النظام قيد التطوير

• Aviasales API - سيتم استخدام البيانات التي يتم إرجاعها بواسطة واجهة برمجة التطبيقات هذه لجميع الأعمال اللاحقة ؛
• مثيل منتج EC2 - جهاز افتراضي منتظم في السحابة سيتم إنشاء دفق بيانات الإدخال عليه:
  
  
  • وكيل Kinesis هو تطبيق Java مثبت محليًا يوفر طريقة سهلة لجمع البيانات وإرسالها إلى Kinesis (Kinesis Data Streams أو Kinesis Firehose). يراقب الوكيل باستمرار مجموعة من الملفات في الدلائل المحددة ويرسل بيانات جديدة إلى Kinesis ؛
  • Caller API Script - نص برمجي Python يجعل طلبات API ويضع الاستجابة في مجلد يراقبه عامل Kinesis ؛
• 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 وإرسال الإشعارات عبر الرسائل القصيرة لا يتم تضمينهما في المستوى المجاني السنوي (الاستخدام المجاني) . ولكن على الرغم من ذلك ، مع أخذ دولارين في الاعتبار ، فمن الممكن تمامًا بناء النظام المقترح واللعب به. وبالطبع ، لا تنس حذف جميع الموارد بعد أن تصبح غير ضرورية.

نظام النشر اليدوي

إعداد تدفقات بيانات Kinesis

إعداد المنتج

إنشاء أدوار IAM لـ EC2

, , 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

1. يسمح لك بإنشاء تدفقات جديدة (دفق الإخراج) بناءً على الاستعلامات الخاصة ببيانات المصدر ؛
2. يوفر دفق مع الأخطاء التي حدثت أثناء عملية التطبيق (دفق الخطأ) ؛
3. يمكنه تحديد مخطط بيانات الإدخال تلقائيًا (يمكن إعادة تعريفه يدويًا إذا لزم الأمر).

هذه خدمة باهظة الثمن - 0.11 دولار أمريكي للساعة ، لذا يجب عليك استخدامها بعناية وإزالتها عند الانتهاء من العمل.

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) و "المضخات" (PUMP) - طلبات الإدراج المستمر التي تدرج البيانات من دفق واحد في تطبيق في دفق آخر.

إنشاء والاشتراك في موضوع SNS

إنشاء جدول في DynamoDB

إنشاء وظيفة جامع لامدا

إنشاء دور 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

استنتاج