🥇Aviasales API integreerimine Amazon Kinesis ja lihtsus serverita

Tere, Habr!

Kas teile meeldib lennata lennukitega? Mulle väga meeldib, kuid isoleerimise ajal hakkasin ka analüüsima andmeid ühe tuntud lennupiletite ressursi — Aviasales — kohta.

Täna käsitleme Amazon Kinesis'i tööd, ehitame reaalajas analüüsiga voogedastussüsteemi, paigaldame NoSQL andmebaasi Amazon DynamoDB põhiliseks andmehulgaks ning seadistame SMS-teavituse huvitavate piletite kohta.

Kõik üksikasjad on allpool! Alustame!

Sissejuhatus

Kuna näide vajab meil juurdepääsu API Aviasales. Juurdepääs sellele on tasuta ja piiranguteta, peate lihtsalt registreeruma jaama "Arendajatele", et saada oma API token andmete juurde pääsemiseks.

Selle artikli peamine eesmärk on anda ülevaade teabe voogedastuse kasutamisest AWS-is, jättes kõrvale, et API kaudu saadud andmed ei ole rangelt ajakohased ja need edastatakse vahemälust, mis moodustatakse Aviasales.ru ja Jetradar.com kasutajate otsingute alusel viimase 48 tunni jooksul.

Kinesis-agendi kaudu saadud lennupiletite andmed, mis on installitud tootmismasinasse, töötlevad automaatselt ja edastavad vajaliku voogu läbi Kinesis Data Analytics. Töötlemata versioon sellest voogust kirjutatakse otse ladustamisse. DynamoDB-s välja töötatud 'toore' andmete ladustamine võimaldab süvitsi analüüsida pileteid BI tööriistade, nagu AWS Quick Sight, kaudu.

Käsitleme kahte infrastruktuuri juurutamise varianti:

Käsitsi — läbi AWS Management Console;
Terraformi koodist infrastruktuur — laiskade automatiseerijate jaoks;

Arendatava süsteemi arhitektuur

Kasutatavad komponendid:

Aviasales API — selle API kaudu tagastatavad andmed kasutatakse edasiseks töötamiseks;
EC2 tootmisinstants — tavaline virtuaalne masin pilves, kus genereeritakse sisendvoog:
- Kinesis Agent — see on Java-rakendus, mis installitakse kohalikult masinale ja pakub lihtsat viisi andmete kogumiseks ja saatmiseks Kinesisesse (Kinesis Data Streams või Kinesis Firehose). Agent jälgib pidevalt määratud kaustades asuvaid failide kogumeid ja saadab uusi andmeid Kinesisesse;
- API Caller skript — Python-skripti, mis teeb API päringuid ja salvestab vastuse kausta, mida jälgib Kinesis Agent;
Kinesis Andmevood — reaalajas andmevoogude teenus laia spektriga skaleerimisvõimetega;
Kinesis Analüütika — serverivaba teenus, mis lihtsustab reaalajas andmevoogude analüüsi. Amazon Kinesis Data Analytics konfigureerib rakenduste tööks vajalikud ressursid ja skaala automaatselt, et hallata iga sissetuleva andmehulka;
AWS Lambda — teenus, mis võimaldab koodi käivitada ilma serverite eraldamise ja seadistamiseta. Kõik arvutusvõimsused skaleeruvad automaatselt iga kutsumise puhul;
Amazon DynamoDB — paaride "võti-väärtus" ja dokumentide andmebaas, mis tagab vähem kui 10 ms latentsuse igas mõõtkavas. DynamoDB kasutamisel ei pea jaotama servereid, installima paranduspakette ega haldama neid. DynamoDB skaleerib automaatselt tabeleid, kohandades saadaolevate ressursside hulka ja hoides kõrge jõudluse. Süsteemi haldamise toimingud pole vajalikud;
Amazon SNS — täielikult hallatav sõnumite saatmise teenus ‘väljund – tellija’ (Pub/Sub) mudeli kaudu, mille abil saab isoleerida mikroteenuseid, hajutatud süsteeme ja serverita rakendusi. SNS-i saab kasutada teabe saatmiseks lõppkasutajatele mobiilsete push-teadete, SMS-sõnumite ja e-kirjade kaudu.

Algne ettevalmistus

Andmevoo emuleerimiseks otsustasin kasutada lennupiletite teavet, mida tagastab API Aviasales. Siin on dokumentatsioonis suhteliselt ulatuslik nimekiri erinevatest meetoditest; võtame neist ühe — ‘Kuu hinna kalender’, mis tagastab hinnad iga kuu päeva kohta, rühmitatuna ümberistumiste arvu järgi. Kui otsinguprogrammi kuud ei edastata, tagastatakse teave kuu kohta, mis järgneb praegusele.

Nii et registreerime end, saame oma tokeni.

Allpool on näidis päring:

http://api.travelpayouts.com/v2/prices/month-matrix?currency=rub&origin=LED&destination=HKT&show_to_affiliates=true&token=TOKEN_API

Ülaltoodud API-andmete saamise meetod koos tokeni edastamisega päringus töötab, kuid mulle meeldib rohkem edastada juurdepääs tokeni kaudu päises, seega kasutame skriptis api_caller.py just seda meetodit.

Näidis vastus:

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

Ülaltoodud API vastuse näites on pilet Peterburist Phuketi… Oh, unistamisest pole mõtet…
Kuna ma olen Kazanist ja Phuket on praegu «meie jaoks vaid unistus», otsime pileteid Peterburist Kazani.

Eeldatakse, et teil on juba AWS konto. Tahan kohe rõhutada, et Kinesis ja teavitamine SMS-i kaudu ei kuulu aastase Tasuta kasutustaseme (Free Tier). Kuid isegi sellega arvestades, mõne dollari investeerides on täiesti võimalik üles ehitada pakutud süsteem ja sellega mängida. Ja loomulikult ei tohi unustada kõik ressursid kustutada, kui need enam vajalikud ei ole.

Õnneks on DynamoDb ja lambda-funktsioonid meile tinglikult tasuta, kui jääte kuiste tasuta limiitide sisse. Näiteks DynamoDB puhul: 25 GB salvestusruumi, 25 WCU/RUC ja 100 miljonit päringut. Ja miljon lambda funktsioonide kutsumist kuus.

Süsteemi käsitsi juurutamine

Kinesis Data Streams seadistamine

Liigume Kinesis Data Streams teenusesse ja loome kaks uut voogu, igas ühes shard.

Mis on shard?
Shard on Amazon Kinesis voogude peamine andmeedastuse üksus. Üks shard võimaldab sissetulevate andmete edastust kiirusel 1 MB/s ja väljuvate andmete edastust kiirusel 2 MB/s. Üks shard toetab kuni 1000 PUT kirjet sekundis. Voogude loomisel tuleb määrata soovitud shardide arv. Näiteks võib luua kaheastmelise voo. See voog tagab sissetulevate andmete edastuse kiirusel 2 MB/s ja väljuvate andmete edastuse kiirusel 4 MB/s, toetades kuni 2000 PUT kirjet sekundis.

Mida rohkem shard'e teie voos on, seda suurem on selle läbilaskevõime. Põhimõtteliselt skaleeritakse vooge shardide lisamise kaudu. Kuid mida rohkem shard'e teil on, seda kallim see on. Iga shard maksab 1,5 senti tunnis ja lisaks 1,4 senti iga miljoni PUT toimingu kohta (PUT payload units).

Loome uue voo nimega airline_tickets, sellele piisab täiesti 1 shardist:

Nüüd loome veel ühe voo nimega special_stream:

Produtsendi seadistamine

Andmete töötluse produtsendina on tavaline EC2 instants ülesande lahendamiseks piisav. See ei pea olema võimas kallis virtuaalmasin, spottide t2.micro sobib täiesti.

Oluline märkus: näidisena tuleks kasutada pilti — Amazon Linux AMI 2018.03.0, sest sellega on Kinesis Agendi kiireks käivitamiseks vähem seadistusi.

Liigume EC2 teenusesse, loome uue virtuaalmasina, valime vajaliku AMI tüübi t2.micro, mis kuulub Free Tieri:

Kuna uuel loodud virtuaalmasinal peab olema õigused Kinesis teenusega suhtlemiseks, tuleb need õigused anda. Parim viis selleks on määrata IAM Roll. Seega tuleb ekraanil Samuti 3: Konfigureeri instantsi üksikasjad valida Loo uus IAM Roll:

IAM rolli loomine EC2 jaoks

Avanenud aknas valime, et loome uue rolli EC2 jaoks ja läheme jaotisse Litsentsid:

Õppenäidisena ei pea me kogu ressursside õiguste detailset seadistamist arvesse võtma, seega valime Amazoni eelkonfigureeritud poliitikad: AmazonKinesisFullAccess ja CloudWatchFullAccess.

Anname sellele rollile mõne mõistliku nime, näiteks: EC2-KinesisStreams-FullAccess. Tulemusena peaks olema sama, mis on näidatud alloleval pildil:

Pärast selle uue rolli loomist ärge unustage seda siduda loodava virtuaalmasina instantsiga:

Sellel ekraanil ei muudeta midagi ja liigutakse järgmistele akendele.

Ketta parameetrid võivad jääda vaikeseadeteks, samuti ka sildid (kuigi hea tava on kasutada silte, andes näiteks instantsile nime ja määrates keskkonna).

Nüüd oleme vahekaartidel Samm 6: Konfigureeri turgroup, kus on vajalik luua uus või näidata olemasolevat turgroup'i, mis lubab SSH kaudu (port 22) instantsiga ühendust võtta. Valige seal Allikas —> Minu IP ja saate instantsi käivitada.

Niipea, kui see läheb staatuseks running, saate proovida sellele SSH kaudu ühendust luua.

Kuna Kinesis Agenti kasutamine on võimaldanud pärast masinaga edukat ühendust, peate terminalis sisestama järgmised käsud:

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

Loomine kaust vastuste salvestamiseks API-le:

sudo mkdir /var/log/airline_tickets

Enne agendi käivitamist on vajalik selle konfiguratsiooni seadistamine:

sudo vim /etc/aws-kinesis/agent.json

Faili agent.json sisu peaks olema järgmine:

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

Nagu näha konfiguratsioonifailist, jälgib agent kataloogis \/var\/log\/airline_tickets\/ faile, millel on .log laiendus, parsimine ja edastamine voolu airline_tickets.

Taaskäivitage teenus ja veenduge, et see käivitub ja töötab:

sudo service aws-kinesis-agent restart

Nüüd laeme alla Python-skripti, mis küsib andmeid API-lt:

REPO_PATH=https:\/\/raw.githubusercontent.com\/igorgorbenko\/aviasales_kinesis\/master\/producer

wget $REPO_PATH\/api_caller.py -P \/home\/ec2-user\/\nwget $REPO_PATH\/requirements.txt -P \/home\/ec2-user\/\nsudo chmod a+x \/home\/ec2-user\/api_caller.py\nsudo \/usr\/local\/bin\/pip3 install -r \/home\/ec2-user\/requirements.txt

Skript api_caller.py küsib andmeid Aviasalesilt ja salvestab saadud vastuse katalooge, mida skanneerib Kinesis agent. Selle skripti teostus on piisavalt standardne, seal on klass TicketsApi, mis võimaldab asünkroonselt API-d kutsuda. Sellesse klassi edastame päise koos märgiga ja päringu parameetrid:

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 occurred: %s', str(err))
            return response_json


def prepare_request(api_token):
    """Return the headers and query for 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 ')
        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)

Seadistuste ja agendi töökindluse testimiseks teeme testkäivituse skripti api_caller.py:

sudo ./api_caller.py TOKEN

Ja vaatame agendi logides ning Monitoring vahekaardil tulemusi andmevoos airline_tickets:

tail -f /var/log/aws-kinesis-agent/aws-kinesis-agent.log

Nagu näha, kõik töötab ja Kinesis Agent edastab andmeid voosse edukalt. Nüüd seadistame consumer'i.

Kinesis Data Analytics seadistamine

Liigume keskse komponendi juurde — loome uue rakenduse Kinesis Data Analytics'isse nimega kinesis_analytics_airlines_app:

Kinesis Data Analytics võimaldab teha reaalaja andmeanalüüsi Kinesis Streams'is SQL keele abil. See on täielikult automaatne teenus (erinevalt Kinesis Streams'ist), mis:

lubab luua uusi vooge (Output Stream) lähtudes algandmetest päringutest;
pakub voogu vigadest, mis tekkisid rakenduste töö käigus (Error Stream);
oskab automaatselt määrata sisendi andmete skeemi (seda saab vajadusel käsitsi üle määrata).

See ei ole odav teenus — 0.11 USD tunni kohta, seega tuleks seda kasutada ettevaatlikult ja eemaldada pärast töö lõppemist.

Ühendame rakenduse andmeallikaga:

Valime voolu, millega kavatseme ühenduda (airline_tickets):

Järgmisena on vaja lisada uus IAM-rolle, et rakendus saaks voost lugeda ja voogu kirjutada. Selleks ei pea Access permissions plokis midagi muutma:

Nüüd küsime andmeskeemi avastamist voos, selleks vajutame nuppu „Discover schema“. Tulemusena uuendatakse (loodakse uus) IAM-rollen ja käivitatakse andmeskeemi avastamine saadud andmetest voos:

Nüüd tuleb minna SQL-redaktorisse. Selle nupu vajutamisel avaneb aken rakenduse käivitamise küsimusega — valime, mida soovime käivitada:

SQL-redaktori aknasse lisame nii lihtsa päringu ja vajutame Save and Run SQL:

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

Relatsioonilistes andmebaasides töötate tabelitega, kasutades INSERT käske kirje lisamiseks ja SELECT käsku andmete pärimiseks. Amazon Kinesis Data Analytics'is töötate voogudega (STREAM) ja «pumpadega» (PUMP) — pidevad sisestamispäringud, mis edastavad andmeid ühest voost rakenduses teise voogu.

Ülaltoodud SQL-käsus otsitakse Aerofloti pileteid, mille hind on alla viie tuhande rubla. Kõik kirjed, mis vastavad nendele tingimustele, paigutatakse DESTINATION_SQL_STREAM voogu.

Sihtkohas valime voolu special_stream ja rippmenüüst rakenduse voonime DESTINATION_SQL_STREAM:

Kõikide toimingute tulemus peaks olema midagi sarnast allolevale pildile:

SNS teema loomine ja sellele registreerimine

Liigume Simple Notification Service'i teenusesse ja loome seal uue teema nimega Airlines:

Registreerime sellel teemal, kus anname mobiiltelefoni numbri, kuhu saadetakse SMS-teated:

Tabeli loomine DynamoDB-s

Töötlemata andmete salvestamiseks voost airline_tickets loome DynamoDB-s sama nimega tabeli. Peamiseks võtmeks kasutame record_id:

Lambda-funktsiooni loomine collector

Loome lambda-funktsiooni nimega Collector, mille ülesanne on küsida airline_tickets voogu ja, kui seal leitakse uusi kirjeid, sisestada need kirjed DynamoDB tabelisse. Ilmselgelt peab see lambda, peale vaikimisi õiguste, olema loodud Kinesis andmevoo lugemiseks ja DynamoDB-sse kirjutamiseks.

IAM rolli loomine lambda-funktsioonile collector
Alustame uue IAM rolli loomisega lambdast nimega Lambda-TicketsProcessingRole:

Testimiseks sobivad täiesti eelhäälestatud poliitikad AmazonKinesisReadOnlyAccess ja AmazonDynamoDBFullAccess, nagu on näidatud alloleval pildil:

See lambda peab käivituma Kinesis'i käivitamisel, kui uusi kirjeid lisatakse airline_stream voogu, seega peame lisama uue käivitaja:

Jäänud on koodi lisamine ja lambda salvestamine.

"""Voogude analüüsimine ja sisestamine DynamoDB tabelisse."""
import base64
import json
import boto3
from decimal import Decimal

DYNAMO_DB = boto3.resource('dynamodb')
TABLE_NAME = 'airline_tickets'

class TicketsParser:
    """Infoteabe analüüsimine voogudest."""

    def __init__(self, table_name, records):
        """Algmeetod."""
        self.table = DYNAMO_DB.Table(table_name)
        self.json_data = TicketsParser.get_json_data(records)

    @staticmethod
    def get_json_data(records):
        """Tagastab deserialiseeritud andmed voogudest."""
        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):
        """Eeltöödelda json andmed."""
        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):
        """Partiis sisestamine tabelisse."""
        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('On lisatud ', len(self.json_data), 'objekti')

def lambda_handler(event, context):
    """Analüüsi voog ja sisestamine DynamoDB tabelisse."""
    print('Sain sündmuse:', event)
    parser = TicketsParser(TABLE_NAME, event['Records'])
    parser.run()

Lambda-funktsiooni notifier loomine

Teine Lambda-funktsioon, mis jälgib teist voogu (special_stream) ja saadab teate SNS-ile, luuakse sarnaselt. Seega peab see Lambda olema ajaloo lugemise ligipääs Kinesise voogudesse ning saadama teateid antud SNS-teemale, mis edastatakse edasi kõigile selle teema tellijatele (e-post, SMS jne).

IAM rolli loomine
Esiteks loome IAM rolli Lambda-KinesisAlarm selle Lambda jaoks ja seejärel määrame selle rolli loodavale Lambda'le alarm_notifier:

See Lambda peab töötama nuhtluse põhjal, kui uusi kirjeid lisatakse voogu special_stream, seega tuleb nuhtlust seadistada sarnaselt sellele, kuidas me tegime Lambda Collectoriga.

Selle Lambda seadistamise hõlbustamiseks loome uue keskkonnamuutuja — TOPIC_ARN, kuhu paneme Airlines teema ANR-d (Amazon Resource Names):

Ja sisestame Lambda koodi, see pole üldse keeruline:

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='Tere! Olen leidnud huvitavat kraami!',
                           Subject='Lennupiletite alarm')
        print('Alarmiteade on edukalt edastatud')
    except Exception as err:
        print('Eddastamine ebaõnnestus', str(err))

Tundub, et süsteemi käsitsi seadistamine on nüüd lõpetatud. Jäänud on vaid testida ja veenduda, et oleme kõik õigesti seadistanud.

Terraformi koodist silti tõstmine

Vajalik ettevalmistus

Terraform — väga mugav avatud lähtekoodiga tööriist infrastruktuuri käivitamiseks koodist. Sellel on oma süntaks, mida on lihtne omandada, ja palju näiteid, kuidas ja mida käivitada. Redaktoris Atom või Visual Studio Code on palju mugavaid pluginaid, mis lihtsustavad töö tegemist Terraformiga.

Tarkvara saab alla laadida siit. Terraformi kõigi võimaluste põhjalik analüüs ületab selle artikli piire, seetõttu piirdume peamiste punktidega.

Kuidas alustada

Kogu projekti kood asub minu repositooriumis. Kloonime enda repositooriumi. Enne käivitamist tuleb veenduda, et teil on installitud ja seadistatud AWS CLI, kuna Terraform kontrollib autentimist failis ~/.aws/credentials.

Hea praktikana, enne kogu infrastruktuuri silti tõstmist, käivitage käsk plan, et näha, mida Terraform meil praegu pilves loomisel on:

terraform.exe plan

Küsimine telefoninumbri sisestamiseks, et saata sellele teatisi. Selles etapis pole selle sisestamine kohustuslik.

Pärast programmi tööplaani analüüsimist saame alustada ressursside loomist:

terraform.exe apply

Pärast selle käsu saatmist ilmub taas telefoninumbri sisestamise küsimus, kirjutame "yes", kui esitatakse küsimus tegevuste tegeliku täitmise kohta. See võimaldab tõsta kogu infrastruktuuri, teha vajalikud EC2 seaded, käivitada lambda-funktsioone jne.

Kuna kõik ressursid on edukalt loodud Terraformi koodi kaudu, peate minema Kinesis Analytics rakenduse detailidesse (kahjuks ei leidnud ma viisi selle tegemiseks otse koodi kaudu).

Käivitame rakenduse:

Pärast seda tuleb selgelt määrata rakenduse voogude nimi, valides rippmenüüst:

Nüüd on kõik tööks valmis.

Rakenduse testimine

Sõltumata sellest, kuidas te süsteemi juurutate, käsitsi või Terraformi koodi kaudu, töötab see ühtemoodi.

Sisestame SSH kaudu EC2 virtuaalmasinasse, kuhu on installitud Kinesis Agent, ja käivitame skripti api_caller.py

sudo ./api_caller.py TOKEN

Jääb oodata SMS-i teie numbrile:

SMS — sõnum tuleb telefonile praktiliselt 1 minuti jooksul:

Jään ootama, et näha, kas DynamoDB andmebaasis on salvestatud kirjed edasiseks, põhjalikumaks analüüsiks. Tabel airline_tickets sisaldab enam-vähem selliseid andmeid:

Kokkuvõte

Töötamise käigus on loodud veebipõhine andmete töötlemise süsteem Amazon Kinesis alusel. Käsitleti Kinesis Agendi kasutamise võimalusi koos Kinesis Data Streams'i ja reaalajas analüütika Kinesis Analytics'iga SQL käskude abil, samuti Amazon Kinesis'i integreerimist teiste AWS teenustega.

Üksikasjalikult kirjeldatud süsteemi oleme juurutatud kahte moodi: suhteliselt pika käsitsi ja kiire koodi Terraform abil.

Kogu projekti lähtekood on saadaval minu GitHubi hoidlas, soovitan sellega tutvuda.

Olen valmis artiklit arutama, ootan teie kommentaare. Lootes konstruktiivset kriitikat.

Soovin edu!

Allikas: habr.com