Integração da API Aviasales com Amazon Kinesis e simplicidade sem servidor

Introdução

O principal objetivo deste artigo é fornecer uma compreensão geral do uso de informações de streaming na AWS, e deixamos de questionar que os dados retornados pela API usada não são estritamente relevantes e são transferidos do cache, que é gerado com base em pesquisas de usuários dos sites Aviasales.ru e Jetradar.com para últimas 48 horas.

• Manual - por meio do AWS Management Console;
• Infraestrutura do código Terraform - para engenheiros de automação preguiçosos;

Arquitetura do sistema em desenvolvimento

• API Aviasales - os dados retornados por essa API serão usados ​​para todo o trabalho subsequente;
• Instância do produtor EC2 - uma máquina virtual comum na nuvem na qual o fluxo de dados de entrada será gerado:
  
  
  • O Kinesis Agent é um aplicativo Java instalado localmente que fornece uma maneira fácil de coletar e enviar dados para o Kinesis (Kinesis Data Streams ou Kinesis Firehose). O agente monitora constantemente um conjunto de arquivos nos diretórios especificados e envia novos dados ao Kinesis;
  • Script da API do chamador - um script Python que faz solicitações de API e coloca a resposta em uma pasta que o Kinesis Agent monitora;
• Kinesis Data Streams — ;
• Kinesis Analytics — , . Amazon Kinesis Data Analytics ;
• AWS Lambda — , . ;
• Amazon DynamoDB — «‑» , 10 . DynamoDB - , . DynamoDB , . ;
• Amazon SNS — « — » (Pub/Sub), , . SNS push-, SMS- .

Treino inicial

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

Supõe-se que você já tenha uma conta da AWS. Quero prestar atenção especial imediatamente, pois o Kinesis e o envio de notificações por SMS não estão incluídos no nível gratuito anual (uso gratuito) . Mas, apesar disso, tendo alguns dólares em mente, é bem possível construir o sistema proposto e brincar com ele. E, é claro, não esqueça de excluir todos os recursos depois que eles se tornarem desnecessários.

Sistema de implantação manual

Configurando o Kinesis Data Streams

Configuração do Produtor

Criando funções do IAM para 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

Configurando o Kinesis Data Analytics

1. permite criar novos fluxos (fluxo de saída) com base em consultas aos dados de origem;
2. fornece um fluxo com erros que ocorreram durante a operação do aplicativo (fluxo de erro);
3. Ele pode determinar automaticamente o esquema de dados de entrada (pode ser redefinido manualmente, se necessário).

Este é um serviço caro - 0,11 USD por hora, portanto, você deve usá-lo com cuidado e removê-lo quando concluir o trabalho.

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

Nos bancos de dados relacionais, você trabalha com tabelas usando instruções INSERT para adicionar registros e uma instrução SELECT para consultar dados. No Amazon Kinesis Data Analytics, você trabalha com fluxos (STREAM) e "bombas" (PUMP) - solicitações de inserção contínua que inserem dados de um fluxo em um aplicativo em outro fluxo.

Criando e assinando o tópico SNS

Criando uma tabela no DynamoDB

Criando uma função de coletor lambda

Criando uma função do IAM para uma função de coletor 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()

Criando um Notificador de Função Lambda

Criar funções do 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))

Implantar a partir do código Terraform

Preparação necessária

Como começar

terraform.exe plan

terraform.exe apply

Teste de Aplicação

sudo ./api_caller.py TOKEN

Conclusão