Aviasales API integrācija ar Amazon Kinesis un serverless vienkāršība

Sveiki, Habr!

Vai jums patīk lidot ar lidmašīnām? Es dievinu, bet pašizolācijas laikā esmu iepazinies arī ar to, kā analizēt datus par aviobiļetēm no viena slavenā resursa – Aviasales.

Šodien mēs apskatīsim Amazon Kinesis darbu, uzbūvēsim straumēšanas sistēmu ar reāllaika analītiku, uzstādīsim NoSQL datubāzi Amazon DynamoDB kā galveno datu glabātuvi un konfigurēsim paziņojumus caur SMS par interesantajiem biļetēm.

Visi sīkumi zemāk! Sāksim!

Ievads

Piemēram, mums nepieciešams piekļuve API Aviasales. Piekļuve tam tiek nodrošināta bez maksas un bez ierobežojumiem, vienkārši jāpierakstās sadaļā "Izstrādātājiem", lai saņemtu savu API tokenu datu piekļuvei.

Šī raksta galvenais mērķis ir sniegt vispārīgu izpratni par informācijas straumēšanas izmantošanu AWS, atmesot faktu, ka ar API atgrieztie dati nav stingri aktuāli un tiek nodoti no kešatmiņas, kas veidojas, pamatojoties uz lietotāju meklējumiem Aviasales.ru un Jetradar.com pēdējās 48 stundās.

Dati par aviobiļetēm, kas iegūti caur API, Kinesis-agent, kas uzstādīts producenta mašīnā, automātiski parsēs un nodos nepieciešamajā plūsmā caur Kinesis Data Analytics. Neapstrādātā šīs plūsmas versija tiks rakstīta tieši glabātavā. Izveidotā DynamoDB ‘sīko’ datu glabātava ļaus veikt dziļāku biļešu analīzi ar BI rīkiem, piemēram, AWS Quick Sight.

Mēs apskatīsim divus infrastruktūras ievietošanas variantus:

• Rokas — caur AWS Management Console;
• Infrastruktūra no Terraform koda — slinkiem automatizētājiem;

Izstrādātās sistēmas arhitektūra

Izmantotie komponenti:

• Aviasales API — dati, kurus atgriež šis API, tiks izmantoti visai turpmākajai darbībai;
• EC2 Ražotāja instance — parasta virtuālā mašīna mākoņos, kur tiks ģenerēts ieejas datu plūsma:
  • Kinesis Agent — tā ir Java lietotne, kas tiek uzstādīta lokāli uz mašīnas, kas nodrošina vienkāršu veidu, kā vākt un sūtīt datus uz Kinesis (Kinesis Data Streams vai Kinesis Firehose). Aģents pastāvīgi uzrauga failu kopas norādītajās direktorijās un sūta jaunus datus uz Kinesis;
  • API Caller skripts — Python skripts, kas veic pieprasījumus uz API un saglabā atbildi mapē, kuru uzrauga Kinesis Agent;
• Kinesis Data Streams — reāllaika datu straumēšanas pakalpojums ar plašām mērogošanas iespējām;
• Kinesis Analytics — bezserveru pakalpojums, kas atvieglo straumējo datu analīzi reālajā laikā. Amazon Kinesis Data Analytics konfigurē resursus lietotņu darbībai un automātiski mērogojas, lai apstrādātu jebkāda apjoma ienākošos datus;
• AWS Lambda — pakalpojums, kas ļauj izpildīt kodu bez serveru rezervēšanas un konfigurācijas. Visas aprēķinu jaudas automātiski mērogojas katrā izsaukumā;
• Amazon DynamoDB — "at鍵- vērtību" un dokumentu datubāze, kas nodrošina aizturi mazāk par 10 milisekundēm jebkurā mērogā. Lietojot DynamoDB, nav nepieciešams dalīt serverus, uzstādīt uz tiem labojumus vai tos pārvaldīt. DynamoDB automātiski mērogo trīs masu, pielāgojot pieejamo resursu apjomu un saglabājot augstu veiktspēju. Nav nepieciešama sistēmas administrēšana;
• Amazon SNS — pilnībā pārvaldīts ziņojumu sūtīšanas pakalpojums pēc "izdevēja — abonenta" modeļa (Pub/Sub), ar kuru var izolēt mikroservisus, izplatītas sistēmas un bezserveru lietotnes. SNS var izmantot, lai nosūtītu informāciju gala lietotājiem, izmantojot mobilos push paziņojumus, SMS ziņojumus un e-pastus.

Sākotnējā sagatavošana

Datu straumes emulācijai es nolēmu izmantot aviobiļešu informāciju, ko atgriež Aviasales API. Tajā dokumentācijā ir diezgan plašs dažādu metožu saraksts; izvēlēsimies vienu no tām — "Cena kalendārs mēnesim", kas atgriež cenas katrai mēneša dienai, grupētas pēc transfēru skaita. Ja meklēšanas pieprasījumā nenorādīsiet mēnesi, informācija tiks atgriezta par mēnesi, kas seko nākamajam.

Tātad, reģistrējamies un iegūstam savu tokenu.

Zemāk ir pieprasījuma piemērs:

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

Iepriekš aprakstītais veids, kā iegūt datus no API, norādot tokenu pieprasījumā, darbosies, taču man vairāk patīk nodot piekļuves tokenu caur galveni, tāpēc skriptā api_caller.py izmantosim tieši šo metodi.

Atbildes piemērs:

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

Iepriekš minētajā API atbildes piemērā ir parādīta biļete no Sanktpēterburgas uz Phuketu… Ai, ko tur sapņot…
Tā kā esmu no Kazaņas, bet Phuketa šobrīd ir «mums tikai sapnī», meklēsim biļetes no Sanktpēterburgas uz Kazaņu.

Tiek pieņemts, ka jums jau ir konts AWS. Jāatzīmē, ka Kinesis un paziņojumu nosūtīšana caur SMS nav ietverta gada Free Tier (bezmaksas lietošana). Pat neskatoties uz to, ja prātā iecere ir dažas dolārus, ir pilnīgi iespējams izveidot piedāvāto sistēmu un spēlēties ar to. Un, protams, nevajadzētu aizmirst dzēst visus resursus pēc tam, kad tie kļuvuši nevajadzīgi.

Pateicoties tam, DynamoDb un lambda funkcijas mums būs nosacīti bez maksas, ja iekļausimies mēneša bezmaksas limitos. Piemēram, DynamoDB: 25 GB krātuves, 25 WCU/RCU un 100 miljoni pieprasījumu. Un miljons lambda funkciju izsaukumu mēnesī.

Rokas sistēmas izvietošana

Kinesis Data Streams konfigurēšana

Pāriesim uz Kinesis Data Streams pakalpojumu un izveidosim divus jaunus plūsmu ar vienu šardu katrā.

Kas ir šards?
Šards ir galvenā datu pārsūtīšanas vienība Amazon Kinesis plūsmā. Viens segments nodrošina ienākošo datu pārsūtīšanu ar ātrumu 1 MB/s un iznākšanas datu pārsūtīšanu ar ātrumu 2 MB/s. Viens segments atbalsta līdz 1000 PUT ierakstiem sekundē. Izveidojot datu plūsmu, ir jānorāda nepieciešamais segmentu skaits. Piemēram, var izveidot datu plūsmu ar diviem segmentiem. Šī datu plūsma nodrošinās ienākošo datu pārsūtīšanu ar ātrumu 2 MB/s un iznākšanas datu pārsūtīšanu ar ātrumu 4 MB/s, atbalstot līdz 2000 PUT ierakstiem sekundē.

Jo vairāk šardu ir jūsu plūsmā, jo lielāka ir tās caurlaidspēja. Principā plūsmas tiek mērogotas, pievienojot šardus. Bet, jo vairāk jums ir šardu, jo augstāka būs cena. Katrs šards maksā 1,5 centus stundā, un papildus 1,4 centus par katriem miljonu pievienošanu plūsmā (PUT payload units).

Izveidosim jaunu plūsmu ar nosaukumu airline_tickets, tam pilnībā pietiks ar 1 shardu:

Tagad izveidosim vēl vienu plūsmu ar nosaukumu special_stream:

Producenta konfigurēšana

Kā datu producentu uzdevuma analīzei pietiks ar parastu EC2 instanci. Tam nevajadzētu būt jaudīgai dārgai virtuālajai mašīnai, pietiek ar t2.micro spotu.

Svarīgs norādījums: piemēram, jāizmanto image — Amazon Linux AMI 2018.03.0, ar to ir mazāk iestatījumu, lai ātri palaistu Kinesis Agent.

Pārejam uz EC2 pakalpojumu, izveidojam jaunu virtuālo mašīnu, izvēlamies nepieciešamo AMI ar tipu t2.micro, kas ietilpst Free Tier:

Lai jaunizveidotā virtuālā mašīna varētu mijiedarboties ar Kinesis pakalpojumu, tai nepieciešams piešķirt attiecīgas tiesības. Labākais veids, kā to izdarīt, ir piešķirt IAM lomu. Tādēļ ekrānā, kas attiecas uz 3. soli: Konfigurēt instancu detaļas, jāizvēlas Izveidot jaunu IAM lomu:

IAM lomas izveide EC2

Atvērtajā logā izvēlamies, ka jauno lomu veidojam EC2 un pārejam uz atļauju sadaļu:

Mācību piemēram var neiedziļināties visās granulu iestatījumu niansēs attiecībā uz resursu tiesībām, tādēļ izvēlēsimies iepriekš iestatītās Amazon politikas: AmazonKinesisFullAccess un CloudWatchFullAccess.

Dodam kādu jēgpilnu nosaukumu šai lomai, piemēram: EC2-KinesisStreams-FullAccess. Rezultātā jābūt tādam pašam, kā attēlā zemāk:

Pēc šīs jaunās lomas izveides neaizmirstam pievienot to izveidojamajai virtuālajai mašīnai:

Vairāk šajā ekrānā neko nemainām un pārejam uz nākamajiem logiem.

Cietā diska opcijas var atstāt noklusējumā, arī tagus (lai gan laba prakse ir izmantot tagus, vismaz norādot instancēm nosaukumu un apkārtni).

Tagad esam 6. solī: Konfigurēt Drošības grupu, kur nepieciešams izveidot jaunu vai izvēlēties esošo drošības grupu, kas ļauj veikt savienojumu, izmantojot ssh (ports 22) ar instanci. Izvēlieties avotu → Mans IP un varat palaist instanci.

Līdzko tā pāries statusā running, varat mēģināt pieslēgties tam, izmantojot ssh.

Lai iegūtu iespēju strādāt ar Kinesis aģentu, pēc veiksmīgas pieslēgšanās mašīnai nepieciešams ievadīt šādas komandas terminālī:

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

Izveidosim mapi API atbilžu glabāšanai:

sudo mkdir /var/log/airline_tickets

Pirms aģenta palaišanas ir jāconfigurē tā iestatījumi:

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

Faila agent.json saturam jāizskatās šādi:

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

Kā redzams konfigurācijas failā, aģents uzraudzīs direktorijā /var/log/airline_tickets/ failus ar paplašinājumu .log, analizēs tos un nosūtīs plūsmā airline_tickets.

Pārlādējam servisu un pārliecināmies, ka tas ir uzsācies un strādā:

sudo service aws-kinesis-agent restart

Tagad lejupielādēsim Python skriptu, kas pieprasīs datus no API:

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

Skripts api_caller.py pieprasa datus no Aviasales un glabā iegūto atbildi direktorijā, kuru skenē Kinesis aģents. Šī skripta īstenošana ir salīdzinoši standarta, ir klase TicketsApi, kas ļauj asinhroni veikt API pieprasījumus. Šai klasei nododam galvenē virkni ar tokenu un pieprasījuma parametriem:

class TicketsApi:
    """API pieprasījumu klase."""

    def __init__(self, headers):
        """Inicēšanas metode."""
        self.base_url = BASE_URL
        self.headers = headers

    async def get_data(self, data):
        """Saņemt datus no API pieprasījuma."""
        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('Atbildes statuss %s: %s',
                            self.base_url, response.status)
                response_json = await response.json()
            except HTTPError as http_err:
                LOGGER.error('Ak, kaut kas nogāja greizi! HTTP kļūda: %s', str(http_err))
            except Exception as err:
                LOGGER.error('Ak, kaut kas nogāja greizi! Kļūda: %s', str(err))
            return response_json


def prepare_request(api_token):
    """Atgriež galvenes un pieprasījuma datus API pieprasījumam."""
    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():
    """Palaist kodu."""
    if len(sys.argv) != 2:
        print('Lietojums: 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 ir atgriezis %s vienības', len(response['data']))
        try:
            count_rows = log_maker(response)
            LOGGER.info('%s rindas ir saglabātas %s',
                        count_rows,
                        TARGET_FILE)
        except Exception as e:
            LOGGER.error('Ak, pieprasījuma rezultāts netika saglabāts failā. %s',
                         str(e))
    else:
        LOGGER.error('Ak, API pieprasījums neizdevās %s!', response)

Lai pārbaudītu iestatījumu pareizību un aģenta darbību, veiksim testējošu skripta api_caller.py palaišanu:

sudo ./api_caller.py TOKEN

Un aplūkosim rezultātus aģenta žurnālos un Monitoring sadaļā datu plūsmā airline_tickets:

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

Kā redzams, viss darbojas un Kinesis Agent veiksmīgi nosūta datus plūsmā. Tagad iestatīsim patērētāju.

Kinesis Data Analytics iestatīšana

Pāriesim pie centrālā sistēmas komponenta — izveidosim jaunu lietojumprogrammu Kinesis Data Analytics ar nosaukumu kinesis_analytics_airlines_app:

Kinesis Data Analytics ļauj veikt datu analīzi reālajā laikā no Kinesis Streams, izmantojot SQL valodu. Tas ir pilnīgi automātiski mērogojams pakalpojums (atšķirībā no Kinesis Streams), kas:

1. ļauj izveidot jaunus plūsmus (Output Stream) pamatojoties uz pieprasījumiem uz izejas datiem;
2. nodrošina plūsmu ar kļūdām, kas radās lietojumprogrammu darbības laikā (Error Stream);
3. spēj automātiski noteikt ienākošo datu shēmu (to var manuāli pārrakstīt vajadzības gadījumā).

Šis nav lēts pakalpojums — 0.11 USD stundā, tāpēc to vajag lietot uzmanīgi un izdzēst pēc lietošanas.

Pievienosim lietojumprogrammu datu avotam:

Izvēlamies plūsmu, kurai plānojam pieslēgties (airline_tickets):

Turpmāk ir nepieciešams pievienot jaunu IAM lomu, lai lietojumprogramma varētu lasīt no plūsmas un rakstīt plūsmā. Tam pietiek neko nemainīt piekļuves atļauju blokā:

Tagad pieprasīsim datu shēmas atklāšanu plūsmā, lai to izdarītu, nospiežam pogu „Discover schema”. Rezultātā tiks atjaunināta (izveidota jauna) IAM loma un tiks uzsākta datu shēmas atklāšana no datiem, kuri jau tika nosūtīti plūsmā:

Tagad ir nepieciešams pāriet uz SQL redaktoru. Nospiežot uz šīs pogas, parādīsies logs ar jautājumu par lietojumprogrammas palaišanu — izvēlamies, ko vēlamies palaist:

SQL redaktorā ielīmēsim šādu vienkāršu vaicājumu un nospiežam 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';

Relāciju datu bāzēs jūs strādājiet ar tabulām, izmantojot INSERT operātorus, lai pievienotu ierakstus, un SELECT operatoru, lai pieprasītu datus. Amazon Kinesis Data Analytics jūs strādājat ar plūsmām (STREAM) un 'sūkņiem' (PUMP) — nepārtrauktiem ievietošanas pieprasījumiem, kas ievieto datus no vienas plūsmas lietojumprogrammā citā plūsmā.

Iepriekš minētajā SQL vaicājumā notiek Aeroflot biļešu meklēšana, kuru cena ir zem piecām tūkstošiem rubļu. Visi ieraksti, kas atbilst šiem nosacījumiem, tiks ievietoti plūsmā DESTINATION_SQL_STREAM.

blokā Destination izvēlamies straumi special_stream, bet izvelkamajā sarakstā In-application stream name DESTINATION_SQL_STREAM:

Visu manipulāciju rezultātā vajadzētu iznākt kaut kas līdzīgs attēlā zemāk:

Topika SNS izveide un abonēšana

Pārejam uz pakalpojumu Simple Notification Service un tur izveidojam jaunu topiku ar nosaukumu Airlines:

Noformējam abonementu uz šo topiku, kurā norādām mobilā telefona numuru, uz kuru tiks sūtīti SMS paziņojumi:

Tabulas izveide DynamoDB

Lai uzglabātu neapstrādātos datus no straumes airline_tickets, izveidosim tabulu DynamoDB ar tādu pašu nosaukumu. Kā primāro atslēgu izmantosim record_id:

Lāmda funkcijas collector izveide

Izveidosim lāmda funkciju ar nosaukumu Collector, kuras uzdevums būs uzraudzīt straumi airline_tickets un, ja tur parādās jauni ieraksti, ievietot šos ierakstus tabulā DynamoDB. Acīmredzot, papildus noklusējuma tiesībām šai lāmda funkcijai jābūt piekļuvei Kinesis datu straumes lasīšanai un rakstīšanai DynamoDB.

IAM lomas izveide lāmda funkcijai collector
Vispirms izveidosim jaunu IAM lomu lāmda funkcijai ar nosaukumu Lambda-TicketsProcessingRole:

Testa piemēram diezgan derēs iepriekš iestatītās politikas AmazonKinesisReadOnlyAccess un AmazonDynamoDBFullAccess, kā parādīts attēlā zemāk:

Šai lāmda funkcijai jāuzsākas pēc Kinesis trigeri, kad jauni ieraksti iekļaujas straumē airline_stream, tāpēc ir jāpieprasa jauns trigeris:

Vēl jāievieto kods un jāsaglabā lāmda.

"""Datu plūsmas analīze un ievietošana DynamoDB tabulā."""
import base64
import json
import boto3
from decimal import Decimal

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

class TicketsParser:
    """Informācijas analīze no plūsmas."""

    def __init__(self, table_name, records):
        """Inicēšanas metode."""
        self.table = DYNAMO_DB.Table(table_name)
        self.json_data = TicketsParser.get_json_data(records)

    @staticmethod
    def get_json_data(records):
        """Atgriež no serializētus datus no plūsmas."""
        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):
        """Priekšapstrādā json datus."""
        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):
        """Grupu ievietošana tabulā."""
        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('Ievietoti ', len(self.json_data), ' vienumi')

def lambda_handler(event, context):
    """Analizē plūsmu un ievieto DynamoDB tabulā."""
    print('Saņemts pasākums:', event)
    parser = TicketsParser(TABLE_NAME, event['Records'])
    parser.run()

Lambdas funkcijas notifier izveide

Otrā lambda funkcija, kas uzraudzīs otro plūsmu (special_stream) un nosūtīs paziņojumus SNS, tiek izveidota līdzīgi. Tādēļ šai lambda ir jābūt piekļuvei lasīt Kinesis un nosūtīt ziņas noteiktajā SNS tēmā, kas tālāk SNS pakalpojums nosūtīs visiem šīs tēmas abonentiem (e-pasts, SMS utt.).

IAM lomas izveide
Vispirms izveidojam IAM lomu Lambda-KinesisAlarm šai lambdai, un pēc tam piešķiram šo lomu izveidotajai lambdai alarm_notifier:

Šai lambdai jādarbojas, kad jauni ieraksti nonāk plūsmā special_stream, tādēļ ir nepieciešams iestatīt trigeri līdzīgi tam, kā to darījām lambdai Collector.

Lai atvieglotu šīs lambdas iestatīšanu, ieviesīsim jaunu vides mainīgo — TOPIC_ARN, kur ievietojam ANR (Amazon Resource Names) tēmas Airlines:

Un ievietojam lambdas kodu, tas ir pavisam vienkāršs:

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='Sveiki! Esmu atradis interesantu informāciju!',
                           Subject='Lidojumu biļešu brīdinājums')
        print('Brīdinājuma ziņojums ir veiksmīgi piegādāts')
    except Exception as err:
        print('Piegādes neizdošanās', str(err))

Izskatās, ka manuāla sistēmas iestatīšana ir beigusies. Paliek tikai pārbaudīt un pārliecināties, ka viss ir noregulēts pareizi.

Iesniegšana no Terraform koda

Nepieciešamā sagatavošana

Terraform — ļoti ērts open-source rīks infrastruktūras izvēršanai no koda. Tam ir savs sintakse, kuru ir viegli apgūt, un daudz piemēru, kā un ko izvērst. Redaktorā Atom vai Visual Studio Code ir daudz noderīgu spraudņu, kas atvieglo darbu ar Terraform.

Izplatīšanu var lejuplādēt no šejienes. Detalizēta visu Terraform iespēju analīze pārsniedz šī raksta robežas, tāpēc aprobežosimies ar svarīgākajiem punktiem.

Kā palaist

Pilnais projekta kods atrodas manā repozitorijā. Klonējam repozitoriju pie sevis. Pirms startēšanas nepieciešams pārliecināties, ka jums ir uzstādīts un konfigurēts AWS CLI, jo Terraform meklēs akreditācijas datus failā ~\/ .aws\/credentials.

Laba prakse pirms visas infrastruktūras izvēršanas ir palaist komandu plan, lai redzētu, ko Terraform pašlaik radīs mākoņos:

terraform.exe plan

Tiks piedāvāts ievadīt telefona numuru, uz kuru tiks sūtīti paziņojumi. Šajā posmā tā ievade nav obligāta.

Pēc programmas darba plāna analīzes varam sākt izstrādāt resursus:

terraform.exe apply

Pēc šīs komandas nosūtīšanas atkal parādīsies aicinājums ievadīt telefona numuru; ierakstiet «jā», kad tiks uzdots jautājums par faktiskas darbības veikšanu. Tas ļaus izveidot visu infrastruktūru, veikt visas nepieciešamās EC2 konfigurācijas un izvērst lambda funkcijas u.c.

Pēc tam, kad visi resursi ir veiksmīgi izveidoti, izmantojot Terraform kodu, nepieciešams ieiet Kinesis Analytics lietojumprogrammas detaļās (diemžēl es neatradu, kā to izdarīt tieši no koda).

Palaižam lietojumprogrammu:

Pēc tam ir skaidri jānorāda in-application stream name, izvēloties no nolaižamā saraksta:

Tagad viss ir gatavs darbam.

Lietojumprogrammas testēšana

Neatkarīgi no tā, kā jūs esat izvērsta sistēmu - manuāli vai caur Terraform kodu, tā darbosies vienādi.

Pievienojamies pa SSH pie EC2 virtuālās mašīnas, kurā ir uzstādīts Kinesis aģents, un palaist api_caller.py skriptu.

sudo ./api_caller.py TOKEN

Palicis tikai gaidīt SMS uz jūsu numuru:

SMS – ziņojums ierodas telefonā praktiski pēc 1 minūtes:

Palicis pārbaudīt, vai ieraksti ir saglabājušies DynamoDB datu bāzē turpmākai, detalizētākai analīzei. Tabula airline_tickets satur aptuveni šādus datus:

Secinājums

Veicot padarīto darbu, tika izveidota tiešsaistes datu apstrādes sistēma, balstoties uz Amazon Kinesis. Tika apskatīti Kinesis aģenta izmantošanas varianti kopā ar Kinesis Data Streams un reāllaika analīzes iespējas, izmantojot Kinesis Analytics SQL komandas, kā arī Amazon Kinesis mijiedarbība ar citiem AWS pakalpojumiem.

Iepriekš aprakstīto sistēmu mēs izvēršam divos veidos: pietiekami ilgā manuālā un ātrajā no Terraform koda.

Visi projekta avota kodi ir pieejami manā GitHub repozitorijā, aicinu iepazīties ar to.

Ar prieku gatavs apspriest rakstu, gaidu jūsu komentārus. Ceru uz konstruktīvu kritiku.

Veiksmi!

Avots: habr.com