Intégration d'API Aviasales avec Amazon Kinesis et simplicité sans serveur

introduction

L'objectif principal de cet article est de donner une compréhension générale de l'utilisation des informations de streaming dans AWS, nous faisons en sorte que les données renvoyées par l'API utilisée ne soient pas strictement pertinentes et soient transférées à partir du cache, qui est généré à partir des recherches des utilisateurs des sites Aviasales.ru et Jetradar.com pour dernières 48 heures.

• Manuel - via AWS Management Console;
• Infrastructure à partir du code Terraform - pour les ingénieurs d'automatisation paresseux;

Architecture du système en cours de développement

• API Aviasales - les données renvoyées par cette API seront utilisées pour tous les travaux ultérieurs;
• EC2 Producer Instance - une machine virtuelle ordinaire dans le cloud sur laquelle le flux de données d'entrée sera généré:
  
  
  • Kinesis Agent est une application Java installée localement qui fournit un moyen simple de collecter et d'envoyer des données à Kinesis (Kinesis Data Streams ou Kinesis Firehose). L'agent surveille en permanence un ensemble de fichiers dans les répertoires spécifiés et envoie de nouvelles données à Kinesis;
  • Script de l'API de l'appelant - un script Python qui fait des demandes d'API et place la réponse dans un dossier surveillé par l'agent Kinesis;
• Kinesis Data Streams — ;
• Kinesis Analytics — , . Amazon Kinesis Data Analytics ;
• AWS Lambda — , . ;
• Amazon DynamoDB — «‑» , 10 . DynamoDB - , . DynamoDB , . ;
• Amazon SNS — « — » (Pub/Sub), , . SNS push-, SMS- .

Formation initiale

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

Il est supposé que vous disposez déjà d'un compte AWS. Je veux tout de suite faire particulièrement attention à ce que Kinesis et l'envoi de notifications par SMS ne soient pas inclus dans le niveau gratuit annuel (utilisation gratuite) . Mais malgré cela, avec quelques dollars en tête, il est tout à fait possible de construire le système proposé et de jouer avec. Et, bien sûr, n'oubliez pas de supprimer toutes les ressources lorsqu'elles deviennent inutiles.

Système de déploiement manuel

Configuration des flux de données Kinesis

Réglage du producteur

Création de rôles IAM pour 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

Configuration de Kinesis Data Analytics

1. vous permet de créer de nouveaux flux (flux de sortie) en fonction des requêtes sur les données source;
2. fournit un flux avec des erreurs survenues pendant le fonctionnement de l'application (flux d'erreurs);
3. Il peut déterminer automatiquement le schéma de données d'entrée (il peut être redéfini manuellement si nécessaire).

Il s'agit d'un service coûteux - 0,11 USD par heure, vous devez donc l'utiliser avec soin et le retirer lorsque vous avez terminé le travail.

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

Dans les bases de données relationnelles, vous travaillez avec des tables à l'aide d'instructions INSERT pour ajouter des enregistrements et d'une instruction SELECT pour interroger des données. Dans Amazon Kinesis Data Analytics, vous travaillez avec des flux (STREAM) et des «pompes» (PUMP) - des demandes d'insertion continues qui insèrent des données d'un flux d'une application dans un autre flux.

Création et abonnement à la rubrique SNS

Création d'une table dans DynamoDB

Création d'une fonction de collecteur lambda

Création d'un rôle IAM pour une fonction de collecteur lambda

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

Création d'un notificateur de fonction lambda

Créer des rôles 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))

Déployer à partir du code Terraform

Préparation nécessaire

Comment commencer

terraform.exe plan

terraform.exe apply

Test d'application

sudo ./api_caller.py TOKEN

Conclusion