Integrasjon av Aviasales API med Amazon Kinesis og enkelhet i serverless

Hei, Habr!

Liker du å fly med fly? Jeg elsker det, men under selvisolasjonen har jeg også blitt glad i å analysere data om flybilletter fra en kjent ressurs — Aviasales.

I dag skal vi gå gjennom hvordan Amazon Kinesis fungerer, bygge et strømmesystem med sanntidsanalyse, sette opp en NoSQL-database Amazon DynamoDB som hoveddatabasen, og konfigurere varsler via SMS for interessante billetter.

Alle detaljer finner du under! La oss begynne!

Innledning

For eksempel trenger vi tilgang til API Aviasales. Tilgangen er gratis og uten begrensninger; du må bare registrere deg under 'Utviklere' for å få din API-token for tilgang til dataene.

Hovedmålet med denne artikkelen er å gi en generell forståelse av hvordan informasjonsstrømming fungerer i AWS. Vi ser bort fra at dataene som returneres av det brukte APIet ikke nødvendigvis er strengt aktuelle og overføres fra cache, som dannes basert på søkene til brukerne av Aviasales.ru og Jetradar.com de siste 48 timene.

Dataene om flybilletter som ble hentet via APIet, vil Kinesis-agenten som er installert på produsentmaskinen, automatisk analysere og overføre til det riktige strømmen via Kinesis Data Analytics. Den ubehandlede versjonen av denne strømmen vil bli skrevet direkte til lagringsenheten. Den oppsatte DynamoDB-lagringen for 'rå' data vil tillate dypere analyse av billetter gjennom BI-verktøy, for eksempel AWS Quick Sight.

Vi vil se på to alternativer for distribusjon av hele infrastrukturen:

• Manuell — gjennom AWS Management Console;
• Infrastruktur fra Terraform-kode — for late automatiserere;

Arkitekturen til det utviklede systemet

Brukte komponenter:

• Aviasales API — dataene som returneres av dette APIet, vil bli brukt til all påfølgende arbeidsprosess;
• EC2 Producer Instance — en vanlig virtuell maskin i skyen som vil generere den innkommende datastreamen:
  • Kinesis Agent — Dette er et Java-program som installeres lokalt på maskinen, og som gir en enkel måte å samle og sende data til Kinesis (Kinesis Data Streams eller Kinesis Firehose). Agenten overvåker kontinuerlig et sett med filer i angitte kataloger og sender nye data til Kinesis;
  • API Callerskript — et Python-skript som sender forespørsel til APIet og legger svaret i en mappe som Kinesis Agent overvåker;
• Kinesis Data Streams — en sanntidsdatastreamingstjeneste med omfattende skaleringsmuligheter;
• Kinesis Analytics — en serverløs tjeneste som forenkler sanntidsanalyse av datastrømmer. Amazon Kinesis Data Analytics konfigurerer ressursene for applikasjonskjøring og skalerer automatisk for å håndtere hvilke mengder innkommende data som helst;
• AWS Lambda — en tjeneste som lar deg kjøre kode uten behov for å sette opp eller tildele servere. All databehandlingskraft skaleres automatisk for hver forespørsel;
• Amazon DynamoDB — en database med «nøkkel-verdi»-par og dokumenter, som gir en forsinkelse på under 10 millisekunder på ethvert nivå. Med DynamoDB kreves det ingen distribusjon av servere, installasjon av oppdateringer eller vedlikehold. DynamoDB skalerer automatisk tabeller ved å justere mengden tilgjengelige ressurser og opprettholder høy ytelse. Ingen systemadministrasjon kreves;
• Amazon SNS — en fullt administrert meldingssendingstjeneste basert på «publisher-subscriber» (Pub/Sub) modell, som lar deg isolere mikrotjenester, distribuerte systemer og serverløse applikasjoner. SNS kan brukes til å sende informasjon til sluttbrukere via mobile push-varsler, SMS-meldinger og e-post.

Innledende forberedelser

For å emulere datastrømmen har jeg valgt å bruke informasjon om flybilletter som tilveiebringes av API Aviasales. I dokumentasjonen et ganske omfattende utvalg av forskjellige metoder, la oss ta en av dem — «Månedens prisoversikt», som returnerer priser for hver dag i måneden, gruppert etter antall mellomlandinger. Hvis man ikke angir måneden for søket i forespørselen, vil informasjon for måneden etter den nåværende bli returnert.

Så, vi registrerer oss og får vår token.

Eksempel på forespørsel under:

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

Metoden for å hente data fra API med token i forespørselen vil fungere, men jeg foretrekker å sende tilgangstoken via header, så i skriptet api_caller.py vil vi bruke nettopp denne metoden.

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

I eksemplet på API-responsen ovenfor vises en billett fra St. Petersburg til Phuket... Åh, hvorfor drømme...
Siden jeg er fra Kazan, og Phuket nå er "bare en drøm" for oss, la oss se etter billetter fra St. Petersburg til Kazan.

Det forutsettes at du allerede har en konto i AWS. Jeg vil straks påpeke at Kinesis og sending av varsler via SMS ikke er inkludert i det årlige. Free Tier (gratis bruk). Men selv med dette, med tanke på et par dollar, kan man godt bygge det foreslåtte systemet og eksperimentere med det. Og selvfølgelig må man ikke glemme å slette alle ressurser etter at de ikke lenger er nødvendige.

Heldigvis vil DynamoDb og Lambda-funksjoner være betinget gratis for oss, så lenge vi holder oss innen de månedlige gratisgrensene. For eksempel for DynamoDB: 25 GB lagring, 25 WCU/RCU og 100 millioner forespørslene. Og en million kall av lambda-funksjoner per måned.

Manuell distribusjon av systemet

Innstilling av Kinesis Data Streams

La oss gå til Kinesis Data Streams-tjenesten og opprette to nye strømmer med én shard hver.

Hva er en shard?
En shard er den grunnleggende enheten for datatransport i Amazon Kinesis-strøm. Én shard gir en datainngang med en hastighet på 1 MB/s og en utgang med en hastighet på 2 MB/s. Én shard støtter opptil 1000 PUT-forespørsel per sekund. Når du oppretter en datastream, må du angi antall shards du ønsker. For eksempel kan man opprette en datastream med to shards. Denne datastreamen vil gi en datainngang med en hastighet på 2 MB/s og en utgang med en hastighet på 4 MB/s med støtte for opptil 2000 PUT-forespørsel per sekund.

Jo flere shards du har i strømmen, desto større blir båndbredden. Generelt skal strømmer skaleres ved å legge til shards. Men jo flere shards du har, desto høyere blir prisen. Hver shard koster 1,5 cent i timen og i tillegg 1,4 cent for hver million operationer med tillegg i strømmen (PUT payload units).

La oss opprette en ny strøm med navnet airline_tickets, én shard vil være mer enn tilstrekkelig:

Nå skal vi opprette en annen strøm med navnet special_stream:

Innstilling av produsent

Som dataprofil for å løse oppgaven, er det tilstrekkelig å bruke en vanlig EC2-instans. Dette trenger ikke være en kraftig og kostbar virtuell maskin, en t2.micro spot vil fungere helt fint.

Viktig merknad: For eksemplet bør man bruke bildet – Amazon Linux AMI 2018.03.0, da dette krever færre innstillinger for rask oppstart av Kinesis Agent.

Vi går til EC2-tjenesten, oppretter en ny virtuell maskin, velger ønsket AMI med typen t2.micro, som inngår i Free Tier:

For at den nyopprettede virtuelle maskinen skal kunne samhandle med Kinesis-tjenesten, må vi gi den de nødvendige rettighetene. Den beste måten å gjøre dette på er å tildele en IAM role. Så på skjermen Trinn 3: Konfigurer instansdetaljer, velg Opprett ny IAM-rolle:

Oppretting av IAM-rolle for EC2

I det åpne vinduet velger vi at vi oppretter en ny rolle for EC2 og går til delen Rettigheter:

I eksempelet kan vi unngå å gå inn i alle detaljene om finjustering av rettigheter for ressurser, så vi velger de forhåndskonfigurerte policyene fra Amazon: AmazonKinesisFullAccess og CloudWatchFullAccess.

Vi gir denne rollen et meningsfullt navn, for eksempel: EC2-KinesisStreams-FullAccess. Resultatet skal være det samme som vist på bildet nedenfor:

Etter opprettelsen av denne nye rollen, ikke glem å knytte den til den opprettede virtuelle maskinen:

Ikke gjør flere endringer på denne skjermen og gå videre til de neste vinduene.

Diskinnstillingene kan stå som standard, og taggene også (selv om god praksis er å bruke tagger, minst gi instansen et navn og spesifisere miljøet).

Nå er vi på fanen Trinn 6: Konfigurer sikkerhetsgruppe, hvor vi må opprette en ny eller spesifisere en eksisterende sikkerhetsgruppe som tillater tilkobling via ssh (port 22) til instansen. Velg der Kilde -> Min IP, og du kan starte instansen.

Så snart den går over i statusen kjørende, kan du prøve å koble til via ssh.

For å bruke Kinesis-agenten etter en vellykket tilkobling til maskinen, må du skrive inn 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

Vi oppretter en mappe for å lagre API-svarene:

sudo mkdir /var/log/airline_tickets

Før du kjører agenten, må du konfigurere konfigurasjonen:

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

Innholdet i filen agent.json skal ha følgende struktur:

{
  "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 det går frem av konfigurasjonsfilen, vil agenten overvåke filer med .log-utvidelse i katalogen /var/log/airline_tickets/, analysere dem og sende dem til strømmen airline_tickets.

Vi restartet tjenesten og bekrefter at den har startet og fungerer:

sudo service aws-kinesis-agent restart

Nå laster vi ned Python-skriptet som skal be om data fra APIet:

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 henter data fra Aviasales og lagrer svaret i katalogen som Kinesis-agenten skanner. Implementeringen av dette skriptet er ganske standard, det finnes en klasse TicketsApi, som gjør det mulig å asynkront hente data fra APIet. I denne klassen sender vi inn hodet med token og forespørselparametrene:

class TicketsApi:
    """API-kallerklasse."""

    def __init__(self, headers):
        """Init-metode."""
        self.base_url = BASE_URL
        self.headers = headers

    async def get_data(self, data):
        """Hent data fra API-forspørsel."""
        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('Responsestatus %s: %s',
                            self.base_url, response.status)
                response_json = await response.json()
            except HTTPError as http_err:
                LOGGER.error('Oops! HTTP-feil oppstod: %s', str(http_err))
            except Exception as err:
                LOGGER.error('Oops! En feil oppstod: %s', str(err))
            return response_json


def prepare_request(api_token):
    """Returner hodene og forespørselen for API-anropet."""
    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():
    """Kjør koden."""
    if len(sys.argv) != 2:
        print('Bruk: 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 har returnert %s elementer', len(response['data']))
        try:
            count_rows = log_maker(response)
            LOGGER.info('%s rader har blitt lagret i %s',
                        count_rows,
                        TARGET_FILE)
        except Exception as e:
            LOGGER.error('Oops! Resultatet av forespørselen ble ikke lagret i fil. %s',
                         str(e))
    else:
        LOGGER.error('Oops! API-forespørsel var ikke vellykket %s!', response)

For å teste konfigurasjonen og agentens funksjonalitet, la oss gjøre en testkjøring av skriptet api_caller.py:

sudo ./api_caller.py TOKEN

Og vi ser resultatet av arbeidet i agentens logger og i Monitoring-fanen i datastreamen airline_tickets:

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

Som syns, alt fungerer og Kinesis Agent sender data til strømmen. Nå konfigurerer vi forbrukeren.

Konfigurering av Kinesis Data Analytics

Vi går videre til den sentrale komponenten i hele systemet — oppretter en ny applikasjon i Kinesis Data Analytics med navnet kinesis_analytics_airlines_app:

Kinesis Data Analytics tillater sanntidsdataanalyse fra Kinesis Streams ved hjelp av SQL. Dette er en fullt selvskalende tjeneste (i motsetning til Kinesis Streams), som:

1. lar deg opprette nye strømmer (Output Stream) basert på forespørslene til kildedataene;
2. gir en feilstrøm som inneholder feil som oppsto under kjøringen av applikasjoner (Error Stream);
3. kan automatisk oppdage strukturen til inngangsdataene (den kan overstyres manuelt hvis nødvendig).

Dette er en kostbar tjeneste — 0,11 USD per time, så vær forsiktig med bruken og slett den ved avslutning.

Vi kobler applikasjonen til datakilden:

Vi velger strømmen vi skal koble oss til (airline_tickets):

Deretter må vi legge til en ny IAM-rolle som lar applikasjonen lese fra strømmen og skrive til strømmen. Det er ikke nødvendig å endre noe i blokken med tilgangstillatelser:

Nå ber vi om oppdagelse av dataskjemaet i strømmen, så vi trykker på knappen «Discover schema». Resultatet vil oppdatere (opprette en ny) IAM-rolle og oppdage skjemaet fra dataene som allerede har kommet til strømmen:

Nå er det nødvendig å gå til SQL-redaktøren. Når vi klikker på denne knappen, vises et vindu med spørsmål om å starte applikasjonen — velg hva vi ønsker å starte:

I SQL-redaktøren limer vi inn en enkel forespørsel og klikker på 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';

I relasjonsdatabaser jobber du med tabeller ved hjelp av INSERT-operasjoner for å legge til oppføringer og SELECT-operasjoner for å hente data. I Amazon Kinesis Data Analytics jobber du med strømmer (STREAM) og pumper (PUMP) — kontinuerlige forespørselinnsettinger som setter inn data fra én strøm til en annen i applikasjonen.

I SQL-forespørselen ovenfor skjer det et søk etter Aeroflot-billetter med en kostnad under fem tusen rubler. Alle oppføringer som oppfyller disse betingelsene, vil bli lagt i strømmen DESTINATION_SQL_STREAM.

I blokken Destination velger vi strømmen special_stream, og i nedtrekkslisten In-application stream name DESTINATION_SQL_STREAM:

Som resultat av alle manipulasjoner skal vi ende opp med noe som ligner bildet nedenfor:

Opprettelse og abonnement på SNS-emne

Vi går til tjenesten Simple Notification Service og oppretter et nytt emne med navnet Airlines:

Vi oppretter et abonnement på dette emnet, der vi angir mobilnummeret som SMS-varsler skal sendes til:

Opprettelse av tabell i DynamoDB

For å lagre de rå dataene fra strømmen airline_tickets, skal vi opprette en tabell i DynamoDB med samme navn. Vi bruker record_id som primærnøkkel:

Opprettelse av Lambda-funksjonen collector

Vi lager en Lambda-funksjon kalt Collector, hvis oppgave blir å spørre strømmen airline_tickets, og ved funn av nye oppføringer, sette disse inn i DynamoDB-tabellen. Det er åpenbart at i tillegg til standardrettighetene må denne Lambda-funksjonen ha tilgang til å lese Kinesis-datastrømmen og skrive til DynamoDB.

Opprettelse av IAM-rolle for Lambda-funksjonen collector
Først oppretter vi en ny IAM-rolle for Lambda med navnet Lambda-TicketsProcessingRole:

For dette testeksempelet er de forhåndsinnstilte policyene AmazonKinesisReadOnlyAccess og AmazonDynamoDBFullAccess tilstrekkelige, som vist på bildet nedenfor:

Denne Lambda-funksjonen må kjøres på et utløser fra Kinesis når nye oppføringer kommer inn i strømmen airline_stream, så vi må legge til en ny utløser:

Det gjenstår bare å sette inn koden og lagre Lambda-funksjonen.

"""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()

Oppretting av lambda-funksjonen notifier

Den andre lambda-funksjonen, som vil overvåke den andre strømmen (special_stream) og sende en varsling til SNS, opprettes på samme måte. Dermed må denne lambda ha tilgang til å lese fra Kinesis og sende meldinger til den angitte SNS-topikken, som deretter distribueres av SNS-tjenesten til alle abonnenter av denne topikken (e-post, SMS, osv.).

Oppretting av IAM-rolle
Først oppretter vi IAM-rollen Lambda-KinesisAlarm for denne lambdaen, og deretter tildeler vi denne rollen til opprettet lambda alarm_notifier:

Denne lambdaen skal trigges ved tilføyelse av nye poster i strømmen special_stream, så det er nødvendig å konfigurere triggen på samme måte som vi gjorde for lambdaen Collector.

For å forenkle konfigurasjonen av denne lambdaen, introduserer vi en ny miljøvariabel — TOPIC_ARN, hvor vi setter ARN (Amazon Resource Names) til Airlines-topikken:

Og vi setter inn koden for lambdaen, den er ganske enkel:

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='Hei! Jeg har funnet noe interessant!',
                           Subject='Alarm for flybilletter')
        print('Alarmmeldingen har blitt levert vellykket')
    except Exception as err:
        print('Leveringsfeil', str(err))

Det ser ut til at manuell systemkonfigurasjon er avsluttet. Det gjenstår bare å teste og sørge for at vi har konfigurert alt riktig.

Distribusjon fra Terraform-kode

Nødvendig forberedelse

Terraform — et veldig praktisk open-source verktøy for distribusjon av infrastruktur fra kode. Det har sin egen syntaks som er lett å lære, og mange eksempler på hvordan man distribuerer hva. I Atom- eller Visual Studio Code-editorene er det mange nyttige plugins som gjør arbeidet med Terraform enklere.

Distributøren kan lastes ned herfrå. En detaljert gjennomgang av alle muligheter i Terraform går utenfor rammen av denne artikkelen, så vi begrenser oss til hovedpunktene.

Hvordan starte

Hele prosjektkoden ligger i mitt repository. Klon repositoryen til din egen. Før du starter, må du sørge for at du har installert og konfigurert AWS CLI, siden Terraform vil lete etter legitimations i filen ~/ .aws/credentials.

Det er god praksis før distribusjon av hele infrastrukturen å kjøre kommandoen plan for å se hva Terraform nå skal opprette i skyen:

terraform.exe plan

Du vil bli bedt om å oppgi telefonnummeret for å sende varsler dit. På dette stadiet er det ikke nødvendig å oppgi det.

Etter å ha analysert programmets arbeidsplan, kan vi starte opprettingen av ressurser:

terraform.exe apply

Etter at denne kommandoen er sendt, vil det igjen bli bedt om å oppgi telefonnummer, skriv «yes» når spørsmålet om faktisk utførelse av handlinger vises. Dette vil gjøre det mulig å sette opp hele infrastrukturen, utføre all nødvendig konfigurasjon av EC2, distribuerer lambda-funksjoner osv.

Etter at alle ressurser er opprettet vellykket via Terraform-kode, må du gå inn i detaljene for Kinesis Analytics-applikasjonen (dessverre fant jeg ikke ut hvordan man gjør dette direkte fra koden).

Start applikasjonen:

Deretter må du eksplisitt angi in-application stream-navn, ved å velge fra nedtrekkslisten:

Nå er alt klart til bruk.

Testing av applikasjonens drift

Uansett om du distribuerte systemet manuelt eller via Terraform-kode, vil det fungere likt.

Logg inn via SSH på EC2-virtuell maskin der Kinesis Agent er installert og start skriptet api_caller.py

sudo ./api_caller.py TOKEN

Bare vent på SMS til ditt nummer:

SMS-en blir sendt til telefonen din omtrent ett minutt senere:

Det gjenstår å sjekke om postene er lagret i DynamoDB-databasen for videre, mer detaljert analyse. Tabellen airline_tickets inneholder omtrent slike data:

Konklusjon

I løpet av arbeidet ble det bygget et system for sanntidsbehandling av data basert på Amazon Kinesis. Vi vurderte alternativer for bruk av Kinesis Agent sammen med Kinesis Data Streams og sanntidsanalyse med Kinesis Analytics ved hjelp av SQL-kommandoer, samt interaksjon mellom Amazon Kinesis og andre AWS-tjenester.

Det ovennevnte systemet ble implementert på to måter: en relativt langvarig manuell metode og en raskere fra Terraform-kode.

All kildekode for prosjektet er tilgjengelig i min repository på GitHub, jeg foreslår at du tar en titt.

Jeg ser frem til å diskutere artikkelen, og venter på dine kommentarer. Jeg håper på konstruktiv kritikk.

Jeg ønsker deg lykke til!

Kilde: habr.com