Integrasi Aviasales API dengan Amazon Kinesis dan kesederhanaan tanpa server

pengantar

Tujuan utama artikel ini adalah untuk memberikan pemahaman umum tentang penggunaan informasi streaming di AWS, kami membuatnya keluar dari pertanyaan bahwa data yang dikembalikan oleh API yang digunakan tidak sepenuhnya relevan dan ditransfer dari cache, yang dihasilkan berdasarkan pencarian pengguna situs Aviasales.ru dan Jetradar.com untuk 48 jam terakhir.

• Manual - melalui Konsol Manajemen AWS;
• Infrastruktur dari kode Terraform - untuk insinyur otomatisasi malas;

Arsitektur sistem dalam pengembangan

• Aviasales API - data yang dikembalikan oleh API ini akan digunakan untuk semua pekerjaan selanjutnya;
• Mesin Virtual Produser EC2 - mesin virtual biasa di cloud tempat aliran data input akan dihasilkan:
  
  
  • Agen Kinesis adalah aplikasi Java yang diinstal secara lokal yang menyediakan cara mudah untuk mengumpulkan dan mengirim data ke Kinesis (Kinesis Data Stream atau Kinesis Firehose). Agen terus memantau satu set file di direktori yang ditentukan dan mengirimkan data baru ke Kinesis;
  • Caller API Script - skrip Python yang membuat permintaan API dan menempatkan respons dalam folder yang dipantau oleh Kinesis Agent;
• Kinesis Data Streams — ;
• Kinesis Analytics — , . Amazon Kinesis Data Analytics ;
• AWS Lambda — , . ;
• Amazon DynamoDB — «‑» , 10 . DynamoDB - , . DynamoDB , . ;
• Amazon SNS — « — » (Pub/Sub), , . SNS push-, SMS- .

Pelatihan awal

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

Diasumsikan bahwa Anda sudah memiliki akun AWS. Saya ingin memberikan perhatian khusus segera bahwa Kinesis dan mengirim pemberitahuan melalui SMS tidak termasuk dalam Free Tier tahunan (penggunaan gratis) . Tetapi meskipun demikian, memiliki beberapa dolar dalam pikiran, sangat mungkin untuk membangun sistem yang diusulkan dan bermain dengannya. Dan, tentu saja, jangan lupa untuk menghapus semua sumber daya setelah mereka menjadi tidak perlu.

Sistem penyebaran manual

Menyiapkan Streaming Data Kinesis

Pengaturan Produser

Membuat Peran IAM untuk 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

Menyiapkan Analisis Data Kinesis

1. memungkinkan Anda untuk membuat aliran baru (Output Stream) berdasarkan permintaan ke sumber data;
2. menyediakan aliran dengan kesalahan yang terjadi selama operasi aplikasi (Stream Kesalahan);
3. Secara otomatis dapat menentukan skema input data (dapat secara manual didefinisikan ulang jika perlu).

Ini adalah layanan mahal - 0,11 USD per jam, jadi Anda harus menggunakannya dengan hati-hati dan menghapusnya saat Anda menyelesaikan pekerjaan.

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

Di database relasional, Anda bekerja dengan tabel menggunakan pernyataan INSERT untuk menambahkan catatan dan pernyataan SELECT untuk meminta data. Di Amazon Kinesis Data Analytics, Anda bekerja dengan stream (STREAM) dan "pompa" (PUMP) - permintaan penyisipan berkelanjutan yang memasukkan data dari satu aliran dalam aplikasi ke aliran lain.

Membuat dan berlangganan topik SNS

Membuat tabel di DynamoDB

Membuat fungsi kolektor lambda

Membuat peran IAM untuk fungsi kolektor 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()

Membuat notifikasi fungsi lambda

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

Menyebarkan dari kode Terraform

Persiapan yang diperlukan

Bagaimana memulainya

terraform.exe plan

terraform.exe apply

Pengujian Aplikasi

sudo ./api_caller.py TOKEN

Kesimpulan