Aviasales API-Integration mit Amazon Kinesis und serverlose Einfachheit

Einführung

Der Hauptzweck dieses Artikels besteht darin, ein allgemeines Verständnis der Verwendung von Streaming-Informationen in AWS zu vermitteln. Wir stellen nicht in Frage, dass die von der verwendeten API zurückgegebenen Daten nicht unbedingt relevant sind und aus dem Cache übertragen werden, der auf der Suche nach Benutzern der Websites Aviasales.ru und Jetradar.com für generiert wird letzte 48 Stunden.

• Handbuch - über die AWS Management Console;
• Infrastruktur aus Terraform-Code - für faule Automatisierungsingenieure;

Architektur des in Entwicklung befindlichen Systems

• Aviasales-API - Die von dieser API zurückgegebenen Daten werden für alle nachfolgenden Arbeiten verwendet.
• EC2 Producer Instance - eine reguläre virtuelle Maschine in der Cloud, auf der der Eingabedatenstrom generiert wird:
  
  
  • Kinesis Agent ist eine lokal installierte Java-Anwendung, mit der Daten auf einfache Weise erfasst und an Kinesis gesendet werden können (Kinesis Data Streams oder Kinesis Firehose). Der Agent überwacht ständig eine Reihe von Dateien in den angegebenen Verzeichnissen und sendet neue Daten an Kinesis.
  • Caller-API-Skript - Ein Python-Skript, das API-Anforderungen stellt und die Antwort in einem Ordner ablegt, den Kinesis Agent überwacht.
• Kinesis Data Streams — ;
• Kinesis Analytics — , . Amazon Kinesis Data Analytics ;
• AWS Lambda — , . ;
• Amazon DynamoDB — «‑» , 10 . DynamoDB - , . DynamoDB , . ;
• Amazon SNS — « — » (Pub/Sub), , . SNS push-, SMS- .

Erstausbildung

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

Es wird davon ausgegangen, dass Sie bereits ein AWS-Konto haben. Ich möchte sofort besonders darauf achten, dass Kinesis und das Versenden von Benachrichtigungen per SMS nicht in der jährlichen kostenlosen Stufe (kostenlose Nutzung) enthalten sind . Trotzdem ist es mit ein paar Dollar durchaus möglich, das vorgeschlagene System aufzubauen und damit zu spielen. Und vergessen Sie natürlich nicht, alle Ressourcen zu löschen, wenn sie unnötig werden.

Manuelles Bereitstellungssystem

Einrichten von Kinesis-Datenströmen

Produzenteneinstellung

Erstellen von IAM-Rollen für 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

Einrichten von Kinesis Data Analytics

1. Ermöglicht das Erstellen neuer Streams (Output Stream) basierend auf Abfragen der Quelldaten.
2. stellt einen Stream mit Fehlern bereit, die während des Anwendungsbetriebs aufgetreten sind (Error Stream);
3. Es kann das Eingabedatenschema automatisch bestimmen (es kann bei Bedarf manuell neu definiert werden).

Dies ist ein teurer Service - 0,11 USD pro Stunde. Sie sollten ihn daher vorsichtig verwenden und entfernen, wenn Sie die Arbeit abgeschlossen haben.

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';

In relationalen Datenbanken arbeiten Sie mit Tabellen, indem Sie INSERT-Anweisungen verwenden, um Datensätze hinzuzufügen, und eine SELECT-Anweisung, um Daten abzufragen. In Amazon Kinesis Data Analytics arbeiten Sie mit Streams (STREAM) und „Pumps“ (PUMP) - fortlaufende Einfügeanforderungen, mit denen Daten aus einem Stream in einer Anwendung in einen anderen Stream eingefügt werden.

Erstellen und Abonnieren des SNS-Themas

Erstellen einer Tabelle in DynamoDB

Erstellen einer Lambda-Kollektorfunktion

Erstellen einer IAM-Rolle für eine Lambda-Kollektorfunktion

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

Erstellen eines Lambda-Funktionsbenachrichtigers

Erstellen Sie IAM-Rollen

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

Bereitstellung aus Terraform-Code

Notwendige Vorbereitung

Wie man anfängt

terraform.exe plan

terraform.exe apply

Anwendungstests

sudo ./api_caller.py TOKEN

Fazit