Amazon Kinesis 및 서버리스 단순성과 Aviasales API 통합

헤이 하브르!

비행기 조종을 좋아하시나요? 나는 그것을 좋아하지만 자가 격리 중에 잘 알려진 리소스인 Aviasales에서 항공권 데이터를 분석하는 것도 좋아했습니다.

오늘은 Amazon Kinesis의 작업을 분석하고, 실시간 분석 기능을 갖춘 스트리밍 시스템을 구축하고, Amazon DynamoDB NoSQL 데이터베이스를 기본 데이터 스토리지로 설치하고, 흥미로운 티켓에 대한 SMS 알림을 설정해 보겠습니다.

모든 세부 사항은 컷 아래에 있습니다! 가다!

소개

예를 들어 다음 항목에 액세스해야 합니다. 항공세일즈 API. 이에 대한 액세스는 제한 없이 무료로 제공됩니다. 데이터에 액세스하려면 API 토큰을 받으려면 "개발자" 섹션에 등록하기만 하면 됩니다.

이 문서의 주요 목적은 AWS에서의 정보 스트리밍 사용에 대한 일반적인 이해를 제공하는 것입니다. 사용된 API에서 반환된 데이터는 엄밀히 말하면 최신이 아니며 캐시에서 전송된다는 점을 고려합니다. 지난 48시간 동안 Aviasales.ru 및 Jetradar.com 사이트 사용자의 검색을 기반으로 형성되었습니다.

API를 통해 수신된 생산 머신에 설치된 Kinesis-agent는 Kinesis Data Analytics를 통해 자동으로 데이터를 구문 분석하고 원하는 스트림으로 전송합니다. 이 스트림의 원시 버전은 저장소에 직접 기록됩니다. DynamoDB에 배포된 원시 데이터 스토리지를 사용하면 AWS Quick Sight와 같은 BI 도구를 통해 더 심층적인 티켓 분석이 가능합니다.

전체 인프라를 배포하기 위한 두 가지 옵션을 고려해 보겠습니다.

• 수동 - AWS Management Console을 통해;
• Terraform 코드의 인프라는 게으른 자동화를 위한 것입니다.

개발된 시스템의 아키텍처

사용된 구성 요소:

• 항공세일즈 API — 이 API에서 반환된 데이터는 모든 후속 작업에 사용됩니다.
• EC2 생산자 인스턴스 — 입력 데이터 스트림이 생성될 클라우드의 일반 가상 머신:
  • 키네시스 에이전트 데이터를 수집하여 Kinesis(Kinesis Data Streams 또는 Kinesis Firehose)로 보내는 쉬운 방법을 제공하는 머신에 로컬로 설치된 Java 애플리케이션입니다. 에이전트는 지정된 디렉터리의 파일 집합을 지속적으로 모니터링하고 새 데이터를 Kinesis로 보냅니다.
  • API 호출자 스크립트 — API에 요청하고 Kinesis 에이전트가 모니터링하는 폴더에 응답을 저장하는 Python 스크립트입니다.
• Kinesis 데이터 스트림 — 광범위한 확장 기능을 갖춘 실시간 데이터 스트리밍 서비스
• 키네시스 분석 스트리밍 데이터 분석을 실시간으로 단순화하는 서버리스 서비스입니다. Amazon Kinesis Data Analytics는 애플리케이션 리소스를 구성하고 모든 양의 수신 데이터를 처리할 수 있도록 자동으로 확장합니다.
• AWS 람다 — 서버를 백업하거나 설정하지 않고도 코드를 실행할 수 있는 서비스입니다. 모든 컴퓨팅 성능은 각 호출에 대해 자동으로 확장됩니다.
• 아마존 DynamoDB - 어떤 규모에서든 실행 시 10밀리초 미만의 대기 시간을 제공하는 키-값 쌍 및 문서의 데이터베이스입니다. DynamoDB를 사용하면 서버를 프로비저닝, 패치 또는 관리할 필요가 없습니다. DynamoDB는 자동으로 테이블을 확장하여 사용 가능한 리소스의 양을 조정하고 높은 성능을 유지합니다. 시스템 관리가 필요하지 않습니다.
• 아마존 SNS - 게시자-구독자(Pub/Sub) 모델을 사용하여 메시지를 보내는 완전 관리형 서비스로, 이를 통해 마이크로서비스, 분산 시스템 및 서버리스 애플리케이션을 격리할 수 있습니다. SNS는 모바일 푸시 알림, SMS 메시지, 이메일을 통해 최종 사용자에게 정보를 보내는 데 사용될 수 있습니다.

초기 교육

데이터 흐름을 에뮬레이트하기 위해 Aviasales API에서 반환된 항공권 정보를 사용하기로 결정했습니다. 안에 선적 서류 비치 매우 광범위한 다양한 방법 목록이 있으므로 그 중 하나인 "월간 가격 달력"을 선택해 보겠습니다. 이 달력은 해당 월의 각 날짜에 대한 가격을 전송 횟수별로 그룹화하여 반환합니다. 요청 시 검색 월을 지정하지 않으면 현재 검색 월의 다음 달에 대한 정보가 반환됩니다.

그럼 등록하고 토큰을 받아봅시다.

요청 예시는 다음과 같습니다.

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

요청에 토큰을 지정하여 API에서 데이터를 수신하는 위의 방법이 작동하지만 헤더를 통해 액세스 토큰을 전달하는 것을 선호하므로 api_caller.py 스크립트에서 이 방법을 사용하겠습니다.

답변 예시:

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

위의 예제 API 응답은 상트페테르부르크에서 푸크(Phuk)까지의 티켓을 보여줍니다. 아, 꿈이군요...
저는 카잔 출신이고 푸켓은 이제 "꿈일 뿐"이므로 상트 페테르부르크에서 카잔까지의 티켓을 찾아 보겠습니다.

이미 AWS 계정이 있다고 가정합니다. Kinesis 및 SMS를 통한 알림 전송은 연간 요금에 포함되지 않는다는 사실에 즉시 특별한 관심을 기울이고 싶습니다. 프리 티어(무료 사용). 그러나 그럼에도 불구하고 몇 달러를 염두에 두고 제안된 시스템을 구축하고 활용하는 것이 가능합니다. 물론, 더 이상 필요하지 않은 모든 리소스를 삭제하는 것을 잊지 마세요.

다행히 월별 무료 한도를 충족하면 DynamoDb 및 Lambda 함수가 무료로 제공됩니다. 예를 들어 DynamoDB의 경우 스토리지 25GB, WCU/RCU 25개, 쿼리 100억 개입니다. 매월 백만 건의 람다 함수 호출이 가능합니다.

수동 시스템 배포

Kinesis Data Streams 설정

Kinesis Data Streams 서비스로 이동하여 각각 샤드 하나씩, 두 개의 새로운 스트림을 생성해 보겠습니다.

샤드란 무엇입니까?
샤드는 Amazon Kinesis 스트림의 기본 데이터 전송 단위입니다. 한 세그먼트는 1MB/s의 속도로 입력 데이터 전송을 제공하고 2MB/s의 속도로 출력 데이터 전송을 제공합니다. 하나의 세그먼트는 초당 최대 1000개의 PUT 항목을 지원합니다. 데이터 스트림을 생성할 때 필요한 세그먼트 수를 지정해야 합니다. 예를 들어 두 개의 세그먼트가 있는 데이터 스트림을 생성할 수 있습니다. 이 데이터 스트림은 2MB/s의 입력 데이터 전송과 4MB/s의 출력 데이터 전송을 제공하며 초당 최대 2000개의 PUT 레코드를 지원합니다.

스트림에 샤드가 많을수록 처리량이 늘어납니다. 원칙적으로 이는 샤드를 추가하여 흐름을 확장하는 방법입니다. 하지만 샤드가 많을수록 가격이 높아집니다. 각 샤드의 비용은 시간당 1,5센트이며, PUT 페이로드 단위 백만 개당 1.4센트가 추가됩니다.

이름으로 새 스트림을 만들어 보겠습니다. 항공권, 1개의 샤드이면 충분합니다.

이제 이름이 다른 스레드를 만들어 보겠습니다. 특수 스트림:

생산자 설정

작업을 분석하려면 일반 EC2 인스턴스를 데이터 생산자로 사용하면 충분합니다. 강력하고 값비싼 가상 머신일 필요는 없습니다. 스팟 t2.micro가 괜찮을 것입니다.

중요 참고 사항: 예를 들어 이미지(Amazon Linux AMI 2018.03.0)를 사용해야 하며 Kinesis 에이전트를 빠르게 시작하기 위한 설정이 더 적습니다.

EC2 서비스로 이동하여 새 가상 머신을 생성하고 프리 티어에 포함된 t2.micro 유형의 원하는 AMI를 선택합니다.

새로 생성된 가상 머신이 Kinesis 서비스와 상호 작용할 수 있으려면 그렇게 할 수 있는 권한이 부여되어야 합니다. 이를 수행하는 가장 좋은 방법은 IAM 역할을 할당하는 것입니다. 따라서 3단계: 인스턴스 세부 정보 구성 화면에서 다음을 선택해야 합니다. 새 IAM 역할 생성:

EC2에 대한 IAM 역할 생성

열리는 창에서 EC2에 대한 새 역할을 생성 중임을 선택하고 권한 섹션으로 이동합니다.

훈련 예제를 사용하면 리소스 권한을 세부적으로 구성하는 복잡한 과정을 모두 살펴볼 필요가 없으므로 Amazon에서 사전 구성된 정책인 AmazonKinesisFullAccess 및 CloudWatchFullAccess를 선택하겠습니다.

이 역할에 의미 있는 이름을 지정해 보겠습니다(예: EC2-KinesisStreams-FullAccess). 결과는 아래 그림과 같아야 합니다.

이 새 역할을 생성한 후에는 생성된 가상 머신 인스턴스에 연결하는 것을 잊지 마세요.

이 화면에서는 아무것도 변경하지 않고 다음 창으로 넘어갑니다.

하드 드라이브 설정과 태그를 기본값으로 둘 수 있습니다(태그를 사용하는 것이 좋지만 최소한 인스턴스에 이름을 지정하고 환경을 표시하는 것이 좋습니다).

이제 6단계: 보안 그룹 구성 탭에 있습니다. 여기서 새 보안 그룹을 생성하거나 기존 보안 그룹을 지정해야 합니다. 이를 통해 SSH(포트 22)를 통해 인스턴스에 연결할 수 있습니다. 소스 -> 내 IP를 선택하면 인스턴스를 시작할 수 있습니다.

실행 상태로 전환되자마자 SSH를 통해 연결을 시도할 수 있습니다.

Kinesis 에이전트로 작업하려면 머신에 성공적으로 연결한 후 터미널에 다음 명령을 입력해야 합니다.

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

API 응답을 저장할 폴더를 만들어 보겠습니다.

sudo mkdir /var/log/airline_tickets

에이전트를 시작하기 전에 해당 구성을 구성해야 합니다.

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

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

구성 파일에서 볼 수 있듯이 에이전트는 /var/log/airline_tickets/ 디렉터리에서 .log 확장자를 가진 파일을 모니터링하고 구문 분석한 후 Airline_tickets 스트림으로 전송합니다.

서비스를 다시 시작하고 서비스가 실행되고 있는지 확인합니다.

sudo service aws-kinesis-agent restart

이제 API에서 데이터를 요청하는 Python 스크립트를 다운로드해 보겠습니다.

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

api_caller.py 스크립트는 Aviasales에서 데이터를 요청하고 Kinesis 에이전트가 검색하는 디렉터리에 수신된 응답을 저장합니다. 이 스크립트의 구현은 매우 표준적이며 TicketsApi 클래스가 있으며 이를 통해 API를 비동기식으로 가져올 수 있습니다. 토큰과 요청 매개변수가 포함된 헤더를 이 클래스에 전달합니다.

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)

에이전트의 올바른 설정과 기능을 테스트하기 위해 api_caller.py 스크립트를 테스트해 보겠습니다.

sudo ./api_caller.py TOKEN

그리고 Agent 로그와 Airlines_tickets 데이터 스트림의 모니터링 탭에서 작업 결과를 살펴봅니다.

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

보시다시피 모든 것이 작동하고 Kinesis 에이전트가 성공적으로 데이터를 스트림으로 보냅니다. 이제 소비자를 구성해 보겠습니다.

Kinesis Data Analytics 설정

전체 시스템의 중심 구성 요소로 넘어가서 Kinesis Data Analytics에서 kinesis_analytics_airlines_app이라는 새 애플리케이션을 생성해 보겠습니다.

Kinesis Data Analytics를 사용하면 SQL 언어를 사용하여 Kinesis Streams에서 실시간 데이터 분석을 수행할 수 있습니다. Kinesis Streams와 달리 다음과 같은 완전 자동 조정 서비스입니다.

1. 소스 데이터에 대한 요청을 기반으로 새로운 스트림(출력 스트림)을 생성할 수 있습니다.
2. 애플리케이션이 실행되는 동안 발생한 오류가 포함된 스트림을 제공합니다(오류 스트림).
3. 입력 데이터 구성표를 자동으로 결정할 수 있습니다(필요한 경우 수동으로 재정의할 수 있음).

작업 시간당 0.11달러라는 저렴한 서비스가 아니므로 주의해서 사용하시고 작업이 끝나면 삭제하시기 바랍니다.

애플리케이션을 데이터 소스에 연결해 보겠습니다.

연결할 스트림(airline_tickets)을 선택합니다.

다음으로, 애플리케이션이 스트림에서 읽고 스트림에 쓸 수 있도록 새 IAM 역할을 연결해야 합니다. 이렇게 하려면 액세스 권한 블록에서 아무것도 변경하지 않는 것으로 충분합니다.

이제 스트림에서 데이터 스키마 검색을 요청해 보겠습니다. 이를 수행하려면 "스키마 검색" 버튼을 클릭하세요. 결과적으로 IAM 역할이 업데이트되고(새 역할이 생성됨) 스트림에 이미 도착한 데이터에서 스키마 감지가 시작됩니다.

이제 SQL 편집기로 이동해야 합니다. 이 버튼을 클릭하면 애플리케이션을 시작할 것인지 묻는 창이 나타납니다. 실행할 항목을 선택하세요.

SQL 편집기 창에 다음과 같은 간단한 쿼리를 삽입하고 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';

관계형 데이터베이스에서는 INSERT 문을 사용하여 레코드를 추가하고 SELECT 문을 사용하여 데이터를 쿼리하는 테이블 작업을 수행합니다. Amazon Kinesis Data Analytics에서는 스트림(STREAM) 및 펌프(PUMP), 즉 애플리케이션의 한 스트림에서 다른 스트림으로 데이터를 삽입하는 지속적인 삽입 요청을 사용합니다.

위에 제시된 SQL 쿼리는 XNUMX루블 미만의 비용으로 Aeroflot 티켓을 검색합니다. 이러한 조건을 충족하는 모든 레코드는 DESTINATION_SQL_STREAM 스트림에 배치됩니다.

대상 블록에서 Special_stream 스트림을 선택하고 인애플리케이션 스트림 이름 DESTINATION_SQL_STREAM 드롭다운 목록에서 다음을 선택합니다.

모든 조작의 결과는 아래 그림과 유사해야 합니다.

SNS 주제 생성 및 구독

단순 알림 서비스로 이동하여 거기에 Airlines라는 이름의 새 주제를 만듭니다.

이 주제를 구독하고 SMS 알림을 보낼 휴대폰 번호를 지정하세요.

DynamoDB에서 테이블 생성

Airlines_tickets 스트림의 원시 데이터를 저장하기 위해 DynamoDB에 동일한 이름의 테이블을 생성해 보겠습니다. Record_id를 기본 키로 사용합니다.

람다 함수 수집기 만들기

Airlines_tickets 스트림을 폴링하고 거기에서 새 레코드가 발견되면 이러한 레코드를 DynamoDB 테이블에 삽입하는 작업을 수행하는 Collector라는 람다 함수를 생성해 보겠습니다. 분명히 기본 권한 외에도 이 람다에는 Kinesis 데이터 스트림에 대한 읽기 액세스 권한과 DynamoDB에 대한 쓰기 액세스 권한이 있어야 합니다.

수집기 람다 함수에 대한 IAM 역할 생성
먼저 Lambda-TicketsProcessingRole이라는 람다에 대한 새 IAM 역할을 생성해 보겠습니다.

테스트 예의 경우 아래 그림과 같이 사전 구성된 AmazonKinesisReadOnlyAccess 및 AmazonDynamoDBFullAccess 정책이 매우 적합합니다.

이 람다는 새 항목이 Airlines_stream에 입력될 때 Kinesis의 트리거에 의해 시작되어야 하므로 새 트리거를 추가해야 합니다.

남은 것은 코드를 삽입하고 람다를 저장하는 것뿐입니다.

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

람다 함수 알리미 만들기

두 번째 스트림(special_stream)을 모니터링하고 SNS에 알림을 보내는 두 번째 람다 함수도 비슷한 방식으로 생성됩니다. 따라서 이 람다에는 Kinesis에서 읽고 특정 SNS 주제에 메시지를 보낼 수 있는 액세스 권한이 있어야 합니다. 그런 다음 메시지는 SNS 서비스에서 이 주제의 모든 구독자(이메일, SMS 등)에게 전송됩니다.

IAM 역할 생성
먼저 이 람다에 대한 IAM 역할 Lambda-KinesisAlarm을 생성한 다음 생성 중인 Alarm_notifier 람다에 이 역할을 할당합니다.

이 람다는 새 레코드가 Special_stream에 입력되는 트리거에서 작동해야 하므로 Collector 람다에서와 동일한 방식으로 트리거를 구성해야 합니다.

이 람다를 더 쉽게 구성할 수 있도록 새로운 환경 변수인 TOPIC_ARN을 도입하겠습니다. 여기에 Airlines 주제의 ANR(Amazon Recourse Names)을 배치합니다.

그리고 람다 코드를 삽입하면 전혀 복잡하지 않습니다.

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 코드에서 배포

필요한 준비

테라 폼 코드에서 인프라를 배포하기 위한 매우 편리한 오픈 소스 도구입니다. 배우기 쉬운 자체 구문이 있으며 배포 방법과 내용에 대한 많은 예가 있습니다. Atom 편집기 또는 Visual Studio Code에는 Terraform 작업을 더 쉽게 해주는 편리한 플러그인이 많이 있습니다.

배포판을 다운로드할 수 있습니다. 따라서. 모든 Terraform 기능에 대한 자세한 분석은 이 기사의 범위를 벗어나므로 주요 사항으로만 제한하겠습니다.

달리는 방법

프로젝트의 전체 코드는 다음과 같습니다. 내 저장소에. 저장소를 우리 자신에게 복제합니다. 시작하기 전에 AWS CLI가 설치 및 구성되어 있는지 확인해야 합니다. 이유는 다음과 같습니다. Terraform은 ~/.aws/credentials 파일에서 자격 증명을 찾습니다.

좋은 방법은 전체 인프라를 배포하기 전에 계획 명령을 실행하여 현재 클라우드에서 Terraform이 생성하고 있는 항목을 확인하는 것입니다.

terraform.exe plan

알림을 보낼 전화번호를 입력하라는 메시지가 표시됩니다. 이 단계에서는 입력할 필요가 없습니다.

프로그램 운영 계획을 분석한 후 리소스 생성을 시작할 수 있습니다.

terraform.exe apply

이 명령을 보낸 후 다시 전화번호를 입력하라는 메시지가 표시되며, 실제 작업 수행에 대한 질문이 표시되면 "예"를 누르십시오. 이를 통해 전체 인프라를 설정하고, EC2에 필요한 모든 구성을 수행하고, 람다 기능을 배포하는 등의 작업을 수행할 수 있습니다.

Terraform 코드를 통해 모든 리소스가 성공적으로 생성된 후에는 Kinesis Analytics 애플리케이션의 세부 정보로 이동해야 합니다(안타깝게도 코드에서 직접 이 작업을 수행하는 방법을 찾지 못했습니다).

애플리케이션을 실행합니다:

그런 다음 드롭다운 목록에서 선택하여 인애플리케이션 스트림 이름을 명시적으로 설정해야 합니다.

이제 모든 것이 준비되었습니다.

애플리케이션 테스트

수동으로 또는 Terraform 코드를 통해 시스템을 배포한 방법에 관계없이 동일하게 작동합니다.

Kinesis 에이전트가 설치된 EC2 가상 머신에 SSH를 통해 로그인하고 api_caller.py 스크립트를 실행합니다.

sudo ./api_caller.py TOKEN

당신이 해야 할 일은 당신의 전화번호로 SMS가 오기를 기다리는 것 뿐입니다:

SMS - 거의 1분 안에 메시지가 전화로 도착합니다.

이후의 보다 자세한 분석을 위해 레코드가 DynamoDB 데이터베이스에 저장되었는지 여부를 확인해야 합니다. Airlines_tickets 테이블에는 대략 다음과 같은 데이터가 포함되어 있습니다.

결론

작업 과정에서 Amazon Kinesis를 기반으로 온라인 데이터 처리 시스템이 구축되었습니다. Kinesis Data Streams 및 SQL 명령을 사용한 실시간 분석 Kinesis Analytics와 함께 Kinesis 에이전트를 사용하는 옵션은 물론 Amazon Kinesis와 다른 AWS 서비스의 상호 작용도 고려되었습니다.

우리는 위 시스템을 두 가지 방법으로 배포했습니다. 다소 긴 수동 방법과 Terraform 코드를 사용한 빠른 방법입니다.

모든 프로젝트 소스 코드를 사용할 수 있습니다 내 GitHub 저장소에, 익숙해지는 것이 좋습니다.

기사에 대해 논의하게 된 것을 기쁘게 생각하며 귀하의 의견을 기대합니다. 건설적인 비판을 바랍니다.

난 당신에게 성공을 기원합니다!

출처 : habr.com