Hei Habr!
Liker du å fly med fly? Jeg elsker det, men under selvisolering ble jeg også forelsket i å analysere data om flybilletter fra en velkjent ressurs - Aviasales.
I dag skal vi analysere arbeidet til Amazon Kinesis, bygge et strømmesystem med sanntidsanalyse, installere Amazon DynamoDB NoSQL-databasen som hoveddatalagring og sette opp SMS-varsler for interessante billetter.
Alle detaljene er under kuttet! Gå!
Innledning
For eksempelet trenger vi tilgang til . Tilgang til den tilbys gratis og uten begrensninger; du trenger bare å registrere deg i "Utviklere"-delen for å motta API-tokenet ditt for å få tilgang til dataene.
Hovedformålet med denne artikkelen er å gi en generell forståelse av bruken av informasjonsstrømming i AWS; vi tar i betraktning at dataene som returneres av API-en som brukes ikke er strengt oppdatert og overføres fra cachen, som er dannet basert på søk fra brukere av Aviasales.ru og Jetradar.com sidene de siste 48 timene.
Kinesis-agent, installert på den produserende maskinen, mottatt via API vil automatisk analysere og overføre data til ønsket strøm via Kinesis Data Analytics. Råversjonen av denne strømmen vil bli skrevet direkte til butikken. Rådatalagringen distribuert i DynamoDB vil tillate dypere billettanalyse gjennom BI-verktøy, for eksempel AWS Quick Sight.
Vi vil vurdere to alternativer for å distribuere hele infrastrukturen:
- Manuell - via AWS Management Console;
- Infrastruktur fra Terraform-kode er for late automater;
Arkitektur av det utviklede systemet
Komponenter som brukes:
- — dataene som returneres av denne API-en vil bli brukt til alt etterfølgende arbeid;
- — en vanlig virtuell maskin i skyen der inndatastrømmen vil bli generert:
- er en Java-applikasjon installert lokalt på maskinen som gir en enkel måte å samle inn og sende data til Kinesis (Kinesis Data Streams eller Kinesis Firehose). Agenten overvåker hele tiden et sett med filer i de angitte katalogene og sender nye data til Kinesis;
- — Et Python-skript som sender forespørsler til API og legger svaret inn i en mappe som overvåkes av Kinesis Agent;
- — sanntidsdatastrømmetjeneste med brede skaleringsmuligheter;
- er en serverløs tjeneste som forenkler analysen av strømmedata i sanntid. Amazon Kinesis Data Analytics konfigurerer applikasjonsressurser og skalerer automatisk for å håndtere alle volumer av innkommende data;
- — en tjeneste som lar deg kjøre kode uten å sikkerhetskopiere eller sette opp servere. All datakraft skaleres automatisk for hver samtale;
- - En database med nøkkelverdi-par og dokumenter som gir ventetid på mindre enn 10 millisekunder når den kjøres i alle skalaer. Når du bruker DynamoDB, trenger du ikke klargjøre, lappe eller administrere noen servere. DynamoDB skalerer automatisk tabeller for å justere mengden tilgjengelige ressurser og opprettholde høy ytelse. Ingen systemadministrasjon er nødvendig;
- - en fullstendig administrert tjeneste for sending av meldinger ved bruk av publisher-subscriber (Pub/Sub)-modellen, som du kan isolere mikrotjenester, distribuerte systemer og serverløse applikasjoner med. SNS kan brukes til å sende informasjon til sluttbrukere gjennom mobile push-varsler, SMS-meldinger og e-poster.
Innledende trening
For å etterligne dataflyten bestemte jeg meg for å bruke flybillettinformasjonen som ble returnert av Aviasales API. I en ganske omfattende liste over forskjellige metoder, la oss ta en av dem - "Månedlig priskalender", som returnerer priser for hver dag i måneden, gruppert etter antall overføringer. Hvis du ikke spesifiserer søkemåneden i forespørselen, vil informasjon bli returnert for måneden etter den gjeldende.
Så la oss registrere deg og få tokenet vårt.
Et eksempel på en forespørsel er nedenfor:
http://api.travelpayouts.com/v2/prices/month-matrix?currency=rub&origin=LED&destination=HKT&show_to_affiliates=true&token=TOKEN_APIMetoden ovenfor for å motta data fra API ved å spesifisere et token i forespørselen vil fungere, men jeg foretrekker å sende tilgangstokenet gjennom overskriften, så vi vil bruke denne metoden i api_caller.py-skriptet.
Eksempel på svar:
{{
"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
}]
}
Eksempel-API-svaret ovenfor viser en billett fra St. Petersburg til Phuk... Å, for en drøm...
Siden jeg er fra Kazan, og Phuket nå er "bare en drøm", la oss se etter billetter fra St. Petersburg til Kazan.
Det forutsetter at du allerede har en AWS-konto. Jeg vil umiddelbart gjøre spesielt oppmerksom på at Kinesis og sending av varsler via SMS ikke er inkludert i den årlige . Men til tross for dette, med et par dollar i tankene, er det fullt mulig å bygge det foreslåtte systemet og leke med det. Og, selvfølgelig, ikke glem å slette alle ressurser etter at de ikke lenger er nødvendige.
Heldigvis vil DynamoDb og lambda-funksjoner være gratis for oss hvis vi oppfyller våre månedlige gratisgrenser. For eksempel for DynamoDB: 25 GB lagringsplass, 25 WCU/RCU og 100 millioner spørringer. Og en million lambdafunksjonsanrop per måned.
Manuell systemimplementering
Sette opp Kinesis datastrømmer
La oss gå til Kinesis Data Streams-tjenesten og lage to nye strømmer, en shard for hver.
Hva er et skjær?
Et shard er den grunnleggende dataoverføringsenheten til en Amazon Kinesis-strøm. Ett segment gir inndataoverføring med en hastighet på 1 MB/s og utdataoverføring med en hastighet på 2 MB/s. Ett segment støtter opptil 1000 PUT-oppføringer per sekund. Når du oppretter en datastrøm, må du spesifisere det nødvendige antallet segmenter. Du kan for eksempel opprette en datastrøm med to segmenter. Denne datastrømmen vil gi inngangsdataoverføring med 2 MB/s og utdataoverføring med 4 MB/s, og støtter opptil 2000 PUT-poster per sekund.
Jo flere shards i strømmen din, desto større er gjennomstrømningen. I prinsippet er det slik strømmer skaleres – ved å legge til skår. Men jo flere skår du har, jo høyere er prisen. Hver shard koster 1,5 cent per time og ytterligere 1.4 cent for hver million PUT nyttelastenheter.
La oss lage en ny strøm med navnet flybiletter, 1 skår vil være nok for ham:
La oss nå lage en ny tråd med navnet spesiell_strøm:
Produsentoppsett
For å analysere en oppgave er det nok å bruke en vanlig EC2-instans som dataprodusent. Det trenger ikke å være en kraftig, dyr virtuell maskin; en spot t2.micro vil gjøre det bra.
Viktig merknad: for eksempel bør du bruke image - Amazon Linux AMI 2018.03.0, den har færre innstillinger for raskt å starte Kinesis Agent.
Gå til EC2-tjenesten, lag en ny virtuell maskin, velg ønsket AMI med typen t2.micro, som er inkludert i Free Tier:
For at den nyopprettede virtuelle maskinen skal kunne samhandle med Kinesis-tjenesten, må den gis rettigheter til det. Den beste måten å gjøre dette på er å tildele en IAM-rolle. På skjermbildet Trinn 3: Konfigurer forekomstdetaljer bør du derfor velge Opprett ny IAM-rolle:
Opprette en IAM-rolle for EC2
I vinduet som åpnes, velg at vi oppretter en ny rolle for EC2 og gå til Tillatelser-delen:
Ved å bruke treningseksemplet trenger vi ikke å gå inn i alle vanskelighetene med granulær konfigurasjon av ressursrettigheter, så vi velger retningslinjene forhåndskonfigurert av Amazon: AmazonKinesisFullAccess og CloudWatchFullAccess.
La oss gi et meningsfullt navn for denne rollen, for eksempel: EC2-KinesisStreams-FullAccess. Resultatet skal være det samme som vist på bildet nedenfor:
Etter å ha opprettet denne nye rollen, ikke glem å knytte den til den opprettede virtuelle maskinforekomsten:
Vi endrer ikke noe annet på denne skjermen og går videre til de neste vinduene.
Harddiskinnstillingene kan stå som standard, så vel som taggene (selv om det er god praksis å bruke tags, gi i det minste forekomsten et navn og angi miljøet).
Nå er vi på trinn 6: Konfigurer sikkerhetsgruppe-fanen, der du må opprette en ny eller spesifisere din eksisterende sikkerhetsgruppe, som lar deg koble til instansen via ssh (port 22). Velg Kilde -> Min IP der, og du kan starte forekomsten.
Så snart den bytter til kjørestatus, kan du prøve å koble til den via ssh.
For å kunne jobbe med Kinesis Agent, etter vellykket tilkobling til maskinen, må du angi følgende kommandoer i terminalen:
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
La oss lage en mappe for å lagre API-svar:
sudo mkdir /var/log/airline_ticketsFør du starter agenten, må du konfigurere konfigurasjonen:
sudo vim /etc/aws-kinesis/agent.jsonInnholdet i agent.json-filen skal se slik ut:
{
"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"]
}
]
}
]
}
Som man kan se fra konfigurasjonsfilen, vil agenten overvåke filer med .log-utvidelsen i /var/log/airline_tickets/-katalogen, analysere dem og overføre dem til airline_tickets-strømmen.
Vi starter tjenesten på nytt og sørger for at den er oppe og går:
sudo service aws-kinesis-agent restartLa oss nå laste ned Python-skriptet som vil be om data fra APIen:
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
Skriptet api_caller.py ber om data fra Aviasales og lagrer det mottatte svaret i katalogen som Kinesis-agenten skanner. Implementeringen av dette skriptet er ganske standard, det er en TicketsApi-klasse, den lar deg asynkront trekke API. Vi sender en overskrift med et token og ber om parametere til denne klassen:
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)
For å teste de riktige innstillingene og funksjonaliteten til agenten, la oss prøvekjøre api_caller.py-skriptet:
sudo ./api_caller.py TOKEN
Og vi ser på resultatet av arbeidet i agentloggene og på Overvåking-fanen i airline_tickets-datastrømmen:
tail -f /var/log/aws-kinesis-agent/aws-kinesis-agent.log
Som du kan se, fungerer alt og Kinesis Agent sender data til strømmen. La oss nå konfigurere forbrukeren.
Sette opp Kinesis Data Analytics
La oss gå videre til den sentrale komponenten i hele systemet – lag en ny applikasjon i Kinesis Data Analytics kalt kinesis_analytics_airlines_app:
Kinesis Data Analytics lar deg utføre sanntidsdataanalyser fra Kinesis Streams ved å bruke SQL-språket. Det er en fullstendig autoskaleringstjeneste (i motsetning til Kinesis Streams) som:
- lar deg lage nye strømmer (Output Stream) basert på forespørsler om å hente data;
- gir en strøm med feil som oppstod mens applikasjoner kjørte (Feilstrøm);
- kan automatisk bestemme inndataskjemaet (det kan omdefineres manuelt om nødvendig).
Dette er ikke en billig tjeneste - 0.11 USD per time arbeid, så du bør bruke den forsiktig og slette den når du er ferdig.
La oss koble applikasjonen til datakilden:
Velg strømmen vi skal koble til (airline_tickets):
Deretter må du legge ved en ny IAM-rolle slik at applikasjonen kan lese fra strømmen og skrive til strømmen. For å gjøre dette er det nok å ikke endre noe i tilgangstillatelsesblokken:
La oss nå be om oppdagelse av dataskjemaet i strømmen; for å gjøre dette, klikk på "Oppdag skjema"-knappen. Som et resultat vil IAM-rollen oppdateres (en ny vil bli opprettet) og skjemadeteksjon vil bli lansert fra dataene som allerede har ankommet strømmen:
Nå må du gå til SQL-editoren. Når du klikker på denne knappen, vises et vindu som ber deg starte programmet - velg hva du vil starte:
Sett inn følgende enkle spørring i SQL-redigeringsvinduet og klikk på Lagre og kjør 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';
I relasjonsdatabaser jobber du med tabeller ved å bruke INSERT-setninger for å legge til poster og en SELECT-setning for å spørre etter data. I Amazon Kinesis Data Analytics jobber du med strømmer (STREAMs) og pumper (PUMPs) – kontinuerlige innsettingsforespørsler som setter inn data fra én strøm i en applikasjon til en annen strøm.
SQL-spørringen presentert ovenfor søker etter Aeroflot-billetter til en pris på under fem tusen rubler. Alle poster som oppfyller disse betingelsene vil bli plassert i DESTINATION_SQL_STREAM-strømmen.
I destinasjonsblokken velger du special_stream-strømmen, og i rullegardinlisten DESTINATION_SQL_STREAM i rullegardinlisten for strømnavn i applikasjonen:
Resultatet av alle manipulasjoner bør være noe som ligner på bildet nedenfor:
Opprette og abonnere på et SNS-emne
Gå til Simple Notification Service og lag et nytt emne der med navnet Airlines:
Abonner på dette emnet og angi mobiltelefonnummeret som SMS-varsler vil bli sendt til:
Lag en tabell i DynamoDB
For å lagre rådataene fra deres flybillettstrøm, la oss lage en tabell i DynamoDB med samme navn. Vi vil bruke record_id som primærnøkkel:
Opprette en lambdafunksjonssamler
La oss lage en lambda-funksjon kalt Collector, hvis oppgave vil være å polle airline_tickets-strømmen og, hvis nye poster blir funnet der, sette inn disse postene i DynamoDB-tabellen. Tydeligvis, i tillegg til standardrettighetene, må denne lambdaen ha lesetilgang til Kinesis-datastrømmen og skrivetilgang til DynamoDB.
Opprette en IAM-rolle for samler-lambda-funksjonen
Først, la oss lage en ny IAM-rolle for lambdaen kalt Lambda-TicketsProcessingRole:
For testeksemplet er de forhåndskonfigurerte AmazonKinesisReadOnlyAccess- og AmazonDynamoDBFullAccess-policyene ganske passende, som vist på bildet nedenfor:
Denne lambdaen skal lanseres av en trigger fra Kinesis når nye oppføringer kommer inn i airline_stream, så vi må legge til en ny trigger:
Det gjenstår bare å sette inn koden og lagre lambdaen.
"""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()
Opprette en lambdafunksjonsvarsler
Den andre lambda-funksjonen, som vil overvåke den andre strømmen (special_stream) og sende en melding til SNS, er opprettet på lignende måte. Derfor må denne lambdaen ha tilgang til å lese fra Kinesis og sende meldinger til et gitt SNS-emne, som så sendes av SNS-tjenesten til alle abonnenter av dette emnet (e-post, SMS, etc.).
Opprette en IAM-rolle
Først oppretter vi IAM-rollen Lambda-KinesisAlarm for denne lambdaen, og tilordner deretter denne rollen til alarm_notifier lambdaen som opprettes:
Denne lambdaen skal fungere på en trigger for at nye poster skal komme inn i special_stream, så du må konfigurere triggeren på samme måte som vi gjorde for Collector lambdaen.
For å gjøre det enklere å konfigurere denne lambdaen, la oss introdusere en ny miljøvariabel - TOPIC_ARN, der vi plasserer ANR (Amazon Recourse Names) for flyselskaper-emnet:
Og sett inn lambdakoden, det er ikke komplisert i det hele tatt:
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))
Det ser ut til at det er her den manuelle systemkonfigurasjonen er fullført. Alt som gjenstår er å teste og sørge for at vi har konfigurert alt riktig.
Distribuer fra Terraform-kode
Nødvendig forberedelse
er et veldig praktisk åpen kildekode-verktøy for å distribuere infrastruktur fra kode. Den har sin egen syntaks som er enkel å lære og har mange eksempler på hvordan og hva som skal distribueres. Atom-editoren eller Visual Studio Code har mange nyttige plugins som gjør arbeidet med Terraform enklere.
Du kan laste ned distribusjonen . En detaljert analyse av alle Terraform-funksjoner er utenfor rammen av denne artikkelen, så vi vil begrense oss til hovedpunktene.
Hvordan starte
Den fullstendige koden for prosjektet er . Vi kloner depotet til oss selv. Før du starter, må du sørge for at du har AWS CLI installert og konfigurert, fordi... Terraform vil se etter legitimasjon i ~/.aws/credentials-filen.
En god praksis er å kjøre plankommandoen før du distribuerer hele infrastrukturen for å se hva Terraform for øyeblikket lager for oss i skyen:
terraform.exe planDu vil bli bedt om å angi et telefonnummer du vil sende varsler til. Det er ikke nødvendig å legge inn det på dette stadiet.
Etter å ha analysert programmets driftsplan, kan vi begynne å lage ressurser:
terraform.exe applyEtter å ha sendt denne kommandoen, vil du igjen bli bedt om å angi et telefonnummer; slå "ja" når et spørsmål om å faktisk utføre handlingene vises. Dette vil tillate deg å sette opp hele infrastrukturen, utføre all nødvendig konfigurasjon av EC2, distribuere lambda-funksjoner, etc.
Etter at alle ressursene har blitt opprettet gjennom Terraform-koden, må du gå inn i detaljene i Kinesis Analytics-applikasjonen (dessverre fant jeg ikke hvordan du gjør dette direkte fra koden).
Start applikasjonen:
Etter dette må du eksplisitt angi navnet på strømmen i applikasjonen ved å velge fra rullegardinlisten:
Nå er alt klart til å gå.
Tester applikasjonen
Uansett hvordan du distribuerte systemet, manuelt eller gjennom Terraform-kode, vil det fungere på samme måte.
Vi logger inn via SSH til den virtuelle EC2-maskinen der Kinesis Agent er installert og kjører api_caller.py-skriptet
sudo ./api_caller.py TOKENAlt du trenger å gjøre er å vente på SMS til nummeret ditt:
SMS - en melding kommer på telefonen din om nesten 1 minutt:
Det gjenstår å se om postene ble lagret i DynamoDB-databasen for senere, mer detaljert analyse. Flybilletttabellen inneholder omtrentlig følgende data:
Konklusjon
I løpet av arbeidet som ble utført, ble det bygget et online databehandlingssystem basert på Amazon Kinesis. Alternativer for bruk av Kinesis Agent i forbindelse med Kinesis Data Streams og sanntidsanalyse Kinesis Analytics ved bruk av SQL-kommandoer, samt interaksjonen mellom Amazon Kinesis og andre AWS-tjenester ble vurdert.
Vi distribuerte systemet ovenfor på to måter: en ganske lang manuell og en rask fra Terraform-koden.
All prosjektkildekode er tilgjengelig , jeg foreslår at du gjør deg kjent med det.
Jeg er glad for å diskutere artikkelen, jeg ser frem til dine kommentarer. Jeg håper på konstruktiv kritikk.
Jeg ønsker deg suksess!
Kilde: www.habr.com