Хеј Хабр!
Да ли волите летење авионима? Волим то, али сам се током самоизолације заљубио и у анализу података о авио картама са једног познатог ресурса – Авиасалес.
Данас ћемо анализирати рад Амазон Кинесис-а, изградити систем за стриминг са аналитиком у реалном времену, инсталирати Амазон ДинамоДБ НоСКЛ базу података као главно складиште података и подесити СМС обавештења за занимљиве карте.
Сви детаљи су испод реза! Иди!
Увод
За пример, потребан нам је приступ . Приступ му је бесплатан и без ограничења; потребно је само да се региструјете у одељку „Програмери“ да бисте добили свој АПИ токен за приступ подацима.
Главна сврха овог чланка је да пружи опште разумевање употребе стримовања информација у АВС-у; узимамо у обзир да подаци које враћа коришћени АПИ нису стриктно ажурирани и да се преносе из кеша, који је формирана на основу претрага корисника сајтова Авиасалес.ру и Јетрадар.цом у последњих 48 сати.
Кинесис-агент, инсталиран на машини за производњу, примљен преко АПИ-ја, аутоматски ће анализирати и пренети податке у жељени ток путем Кинесис Дата Аналитицс. Сирова верзија овог стрима ће бити написана директно у продавницу. Складиште сирових података примењено у ДинамоДБ-у ће омогућити дубљу анализу улазница путем БИ алата, као што је АВС Куицк Сигхт.
Размотрићемо две опције за постављање целокупне инфраструктуре:
- Ручно - преко АВС Манагемент Цонсоле;
- Инфраструктура из Терраформ кода је за лење аутоматоре;
Архитектура развијеног система
Коришћене компоненте:
- — подаци које врати овај АПИ ће се користити за сав наредни рад;
- — обична виртуелна машина у облаку на којој ће се генерисати улазни ток података:
- је Јава апликација инсталирана локално на машини која пружа једноставан начин за прикупљање и слање података у Кинесис (Кинесис Дата Стреамс или Кинесис Фирехосе). Агент стално надгледа скуп датотека у наведеним директоријумима и шаље нове податке у Кинесис;
- — Питхон скрипта која поставља захтеве АПИ-ју и ставља одговор у фасциклу коју надгледа Кинесис Агент;
- — услуга стримовања података у реалном времену са широким могућностима скалирања;
- је услуга без сервера која поједностављује анализу стримовања података у реалном времену. Амазон Кинесис Дата Аналитицс конфигурише ресурсе апликације и аутоматски се скалира за руковање било којом количином долазних података;
- — услуга која вам омогућава да покренете код без прављења резервних копија или подешавања сервера. Сва рачунарска снага се аутоматски скалира за сваки позив;
- – База података парова кључ-вредност и докумената која обезбеђује кашњење мање од 10 милисекунди када се ради на било којој скали. Када користите ДинамоДБ, не морате да обезбеђујете, крпите или управљате серверима. ДинамоДБ аутоматски скалира табеле да би прилагодио количину доступних ресурса и одржао високе перформансе. Није потребна системска администрација;
- је потпуно управљана услуга за размену порука заснована на моделу издавач-претплатник (Пуб/Суб), која се може користити за изоловање микросервиса, дистрибуираних система и апликација без сервера. СНС се може користити за слање информација крајњим корисницима путем мобилних пусх обавештења, СМС порука и е-поште.
Иницијална обука
Да бих емулирао ток података, одлучио сам да користим информације о авионским картама које је вратио Авиасалес АПИ. ИН прилично опсежна листа различитих метода, узмимо један од њих - „Месечни календар цена“, који враћа цене за сваки дан у месецу, груписане по броју трансфера. Ако у захтеву не наведете месец претраге, информације ће бити враћене за месец који следи за текућим.
Дакле, хајде да се региструјемо и добијемо наш токен.
Пример захтева је у наставку:
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
}]
}
Пример АПИ одговора изнад показује карту од Санкт Петербурга до Пука... Ох, какав сан...
Пошто сам ја из Казања, а Пукет је сада „само сан“, хајде да потражимо карте од Санкт Петербурга до Казања.
Претпоставља се да већ имате АВС налог. Посебно бих одмах скренуо пажњу да Кинесис и слање обавештења путем СМС-а нису укључени у годишњи . Али чак и упркос томе, са пар долара на уму, сасвим је могуће изградити предложени систем и играти се са њим. И, наравно, не заборавите да избришете све ресурсе након што више нису потребни.
На срећу, ДинамоДб и ламбда функције ће бити бесплатне за нас ако испунимо наше месечне бесплатне границе. На пример, за ДинамоДБ: 25 ГБ складишта, 25 ВЦУ/РЦУ и 100 милиона упита. И милион позива ламбда функција месечно.
Ручно постављање система
Подешавање Кинесис Дата Стреамс
Идемо на услугу Кинесис Дата Стреамс и направимо два нова тока, по један шард за сваки.
Шта је крхотина?
Шард је основна јединица за пренос података Амазон Кинесис тока. Један сегмент обезбеђује пренос улазних података брзином од 1 МБ/с и пренос излазних података брзином од 2 МБ/с. Један сегмент подржава до 1000 ПУТ уноса у секунди. Када креирате ток података, потребно је да наведете потребан број сегмената. На пример, можете да креирате ток података са два сегмента. Овај ток података ће обезбедити пренос улазних података брзином од 2 МБ/с и пренос излазних података брзином од 4 МБ/с, подржавајући до 2000 ПУТ записа у секунди.
Што је више фрагмената у вашем стреаму, већа је његова пропусност. У принципу, овако се скалирају токови - додавањем крхотина. Али што више комада имате, то је већа цена. Сваки комад кошта 1,5 центи по сату и додатних 1.4 цента за сваки милион ПУТ јединица корисног оптерећења.
Хајде да направимо нови ток са именом авионске карте, биће му довољна 1 крхотина:
Сада направимо другу нит са именом специал_стреам:
Подешавање произвођача
За анализу задатка довољно је користити редовну ЕЦ2 инстанцу као произвођач података. Не мора да буде моћна, скупа виртуелна машина; спот т2.мицро ће бити сасвим добар.
Важна напомена: на пример, требало би да користите слику - Амазон Линук АМИ 2018.03.0, има мање подешавања за брзо покретање Кинесис Агент-а.
Идите на услугу ЕЦ2, креирајте нову виртуелну машину, изаберите жељени АМИ са типом т2.мицро, који је укључен у бесплатни ниво:
Да би новостворена виртуелна машина могла да ступи у интеракцију са услугом Кинесис, мора јој се дати права за то. Најбољи начин да то урадите је да доделите ИАМ улогу. Стога, на екрану Корак 3: Конфигуришите детаље о инстанци, требало би да изаберете Креирајте нову ИАМ улогу:
Креирање ИАМ улоге за ЕЦ2
У прозору који се отвори изаберите да креирамо нову улогу за ЕЦ2 и идите на одељак Дозволе:
Користећи пример обуке, не морамо да улазимо у све замршености грануларне конфигурације права на ресурсе, па ћемо изабрати смернице које је Амазон унапред конфигурисао: АмазонКинесисФуллАццесс и ЦлоудВатцхФуллАццесс.
Хајде да дамо неко смислено име за ову улогу, на пример: ЕЦ2-КинесисСтреамс-ФуллАццесс. Резултат би требао бити исти као што је приказано на слици испод:
Након креирања ове нове улоге, не заборавите да је приложите креираној инстанци виртуелне машине:
Не мењамо ништа друго на овом екрану и прелазимо на следеће прозоре.
Подешавања чврстог диска се могу оставити као подразумевана, као и ознаке (иако је добра пракса да користите ознаке, бар дајте инстанци име и означите окружење).
Сада смо на картици Корак 6: Конфигуришите безбедносну групу, где треба да креирате нову или наведете постојећу безбедносну групу, која вам омогућава да се повежете преко ссх-а (порт 22) на инстанцу. Изаберите Извор -> Мој ИП тамо и можете покренути инстанцу.
Чим се пребаци у статус покретања, можете покушати да се повежете на њега преко ссх-а.
Да бисте могли да радите са Кинесис Агентом, након успешног повезивања на машину, морате унети следеће команде у терминал:
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
Као што видите, све функционише и Кинесис Агент успешно шаље податке у стреам. Сада да конфигуришемо потрошача.
Подешавање Кинесис Дата Аналитицс
Пређимо на централну компоненту читавог система – креирајте нову апликацију у Кинесис Дата Аналитицс под називом кинесис_аналитицс_аирлинес_апп:
Кинесис Дата Аналитицс вам омогућава да извршите анализу података у реалном времену из Кинесис Стреамс-а користећи СКЛ језик. То је услуга потпуно аутоматског скалирања (за разлику од Кинесис Стреамс-а) која:
- омогућава вам да креирате нове токове (Оутпут Стреам) на основу захтева за изворне податке;
- пружа ток са грешкама које су се десиле док су апликације биле покренуте (Еррор Стреам);
- може аутоматски одредити шему улазних података (може се ручно редефинисати ако је потребно).
Ово није јефтина услуга - 0.11 УСД по сату рада, па је треба пажљиво користити и обрисати када завршите.
Хајде да повежемо апликацију са извором података:
Изаберите ток на који ћемо се повезати (аирлине_тицкетс):
Затим морате да приложите нову ИАМ улогу тако да апликација може да чита из стрима и пише у ток. Да бисте то урадили, довољно је да не мењате ништа у блоку Приступне дозволе:
Сада тражимо откривање шеме података у стриму; да бисте то урадили, кликните на дугме „Откриј шему“. Као резултат тога, ИАМ улога ће бити ажурирана (направиће се нова) и покренуће се откривање шеме из података који су већ стигли у ток:
Сада морате да одете у СКЛ едитор. Када кликнете на ово дугме, појавиће се прозор са захтевом да покренете апликацију - изаберите шта желите да покренете:
Уметните следећи једноставан упит у прозор СКЛ едитора и кликните на Сачувај и покрени СКЛ:
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';
У релационим базама података радите са табелама користећи ИНСЕРТ наредбе за додавање записа и СЕЛЕЦТ наредбу за упите података. У Амазон Кинесис Дата Аналитицс радите са токовима (СТРЕАМ-ови) и пумпама (ПУМП-и)—захтевима за континуирано уметање који убацују податке из једног тока у апликацији у други ток.
Горе представљен СКЛ упит тражи карте Аерофлота по цени испод пет хиљада рубаља. Сви записи који испуњавају ове услове биће смештени у ток ДЕСТИНАТИОН_СКЛ_СТРЕАМ.
У блоку Дестинатион изаберите стрим специал_стреам, а у падајућој листи назив тока у апликацији ДЕСТИНАТИОН_СКЛ_СТРЕАМ:
Резултат свих манипулација би требао бити нешто сличан слици испод:
Креирање и претплата на СНС тему
Идите на Симпле Нотифицатион Сервице и тамо направите нову тему са именом Аирлинес:
Претплатите се на ову тему и назначите број мобилног телефона на који ће се слати СМС обавештења:
Направите табелу у ДинамоДБ-у
Да бисмо сачували необрађене податке из њиховог тока аирлине_тицкетс, хајде да направимо табелу у ДинамоДБ са истим именом. Користићемо рецорд_ид као примарни кључ:
Креирање колектора ламбда функција
Хајде да направимо ламбда функцију под називом Цоллецтор, чији ће задатак бити да испита ток аирлине_тицкетс и, ако се тамо пронађу нови записи, убаци ове записе у табелу ДинамоДБ. Очигледно, поред подразумеваних права, ова ламбда мора имати приступ за читање Кинесис тока података и приступ за писање у ДинамоДБ.
Креирање ИАМ улоге за колекторску ламбда функцију
Прво, направимо нову ИАМ улогу за ламбда под називом Ламбда-ТицкетсПроцессингРоле:
За пример теста, унапред конфигурисане политике АмазонКинесисРеадОнлиАццесс и АмазонДинамоДБФуллАццесс су сасвим прикладне, као што је приказано на слици испод:
Ова ламбда би требало да буде покренута окидачем из Кинесиса када нови уноси уђу у аирлине_стреам, тако да морамо да додамо нови окидач:
Остаје само да убаците код и сачувате ламбда.
"""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()
Креирање нотифиер-а ламбда функције
Друга ламбда функција, која ће надгледати други ток (специал_стреам) и слати обавештење СНС-у, креирана је на сличан начин. Дакле, ова ламбда мора имати приступ за читање са Кинесиса и слање порука на дату СНС тему, које ће затим СНС сервис послати свим претплатницима ове теме (е-маил, СМС итд.).
Креирање ИАМ улоге
Прво креирамо ИАМ улогу Ламбда-КинесисАларм за ову ламбда, а затим додељујемо ову улогу ламбди аларм_нотифиер која се креира:
Ова ламбда треба да ради на окидачу за улазак нових записа у специал_стреам, тако да морате да конфигуришете окидач на исти начин као што смо урадили за Цоллецтор ламбда.
Да бисмо олакшали конфигурисање ове ламбда, хајде да уведемо нову променљиву окружења - ТОПИЦ_АРН, где постављамо АНР (Амазон Рецоурсе Намес) теме Аирлинес:
И убаците ламбда код, уопште није компликовано:
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))
Чини се да је ту завршена ручна конфигурација система. Остаје само да тестирамо и уверимо се да смо све исправно конфигурисали.
Поставите из Терраформ кода
Потребна припрема
је веома згодан алат отвореног кода за примену инфраструктуре из кода. Има сопствену синтаксу коју је лако научити и има много примера како и шта да се примени. Атом едитор или Висуал Студио Цоде има много практичних додатака који олакшавају рад са Терраформом.
Можете преузети дистрибуцију . Детаљна анализа свих могућности Терраформа је ван оквира овог чланка, па ћемо се ограничити на главне тачке.
Како се трчи
Пун код пројекта је . Ми клонирамо спремиште за себе. Пре него што почнете, морате да се уверите да имате инсталиран и конфигурисан АВС ЦЛИ, јер... Терраформ ће тражити акредитиве у ~/.авс/цредентиалс датотеци.
Добра пракса је да покренете команду плана пре постављања целокупне инфраструктуре да бисте видели шта Терраформ тренутно ствара за нас у облаку:
terraform.exe planОд вас ће бити затражено да унесете број телефона на који ћете слати обавештења. У овој фази није потребно улазити у њега.
Након анализе плана рада програма, можемо започети креирање ресурса:
terraform.exe applyНакон слања ове команде, од вас ће се поново тражити да унесете број телефона; бирајте „да“ када се прикаже питање о стварном извођењу радњи. Ово ће вам омогућити да поставите целокупну инфраструктуру, извршите сву потребну конфигурацију ЕЦ2, примените ламбда функције итд.
Након што су сви ресурси успешно креирани преко Терраформ кода, потребно је да уђете у детаље апликације Кинесис Аналитицс (нажалост, нисам нашао како да то урадим директно из кода).
Покрените апликацију:
Након овога, морате експлицитно да подесите назив тока у апликацији тако што ћете изабрати са падајуће листе:
Сада је све спремно.
Тестирање апликације
Без обзира на то како сте применили систем, ручно или преко Терраформ кода, он ће радити исто.
Пријављујемо се преко ССХ-а на ЕЦ2 виртуелну машину на којој је инсталиран Кинесис Агент и покрећемо скрипту апи_цаллер.пи
sudo ./api_caller.py TOKENСве што треба да урадите је да сачекате СМС на ваш број:
СМС - порука стиже на телефон за скоро 1 минут:
Остаје да се види да ли су записи сачувани у ДинамоДБ бази података ради накнадне, детаљније анализе. Табела аирлине_тицкетс садржи приближно следеће податке:
Закључак
У току обављеног посла изграђен је онлајн систем за обраду података заснован на Амазон Кинесис. Размотрене су опције за коришћење Кинесис Агента у комбинацији са Кинесис Дата Стреамс-ом и аналитиком у реалном времену Кинесис Аналитицс коришћењем СКЛ команди, као и интеракција Амазон Кинесис-а са другим АВС сервисима.
Горњи систем смо применили на два начина: прилично дугачак ручни и брзи из Терраформ кода.
Сав изворни код пројекта је доступан , предлажем да се упознате са тим.
Драго ми је да расправљам о чланку, радујем се вашим коментарима. Надам се конструктивној критици.
Желим вам успех!
Извор: ввв.хабр.цом