이 첫 번째 장에서는 배치 처리와 스트림 처리의 차이를 배우고, 첫 스트림을 생성한 뒤 권한을 관리하고, 스트림에 쓰고 읽는 방법을 익힙니다.

스트리밍이란 무엇이며 왜 중요할까요?

배치 vs 스트림

프로듀서와 대상지

Firehose 딜리버리 스트림 관리

첫 번째 스트림을 준비해 볼까요

역할 생성하기

IAM 사용자 vs. 역할

권한 연습

S3 버킷 생성

Firehose 전송 스트림 다루기

첫 번째 Firehose 스트림 만들기

Firehose 스트림에 기록하기

Firehose 데이터 읽기

클라우드에서의 스트리밍

스트리밍 데이터 여정의 다음 단계는 변환용 Lambda 함수를 사용해 서버리스로 전환하는 것입니다. 실습을 통해 Lambda 레이어를 추가하고 특정 조건에서 Lambda 함수를 트리거해 보겠습니다.

서버리스로 가기

Lambda 함수 생성 및 실행

Options

Great for a Server

Great for Serverless

Serverless vs servers

Trigger

Handler

Layer

Lambda 함수 구성 요소

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(____, ____):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:____' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3
import json

def record_created_handler(event, context):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:Put' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }                 

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

이벤트에 따라 S3에서 데이터 읽기

첫 번째 라이브 Lambda!

Lambda 레이어 추가하기

Lambda Flow

Lambda 흐름

환경 변수 사용하기

서버리스 데이터 워크플로

옵션

빠르게 실행되는 Lambda

실행에 더 오래 걸리는 Lambda

빠른 Lambda vs 느린 Lambda

시간 트리거 Lambda 만들기

시간 트리거 Lambda

시간 트리거 Lambda 코드

서버리스 API

API 트리거형 Lambda 만들기

API Lambda 만들기

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['____']['async']
    if async == 'true':
        exec_type = '____'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.____(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               

# PLEASE DO NOT EDIT
# aws_wrangler_c.py
def read_csv(path, boto3_session, delimiter):
  bucket = path.split("//")[1].split("/")[0]
  prefix = path.split("//")[1].replace(f"{bucket}/", "")
  objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']
  dfs = []
  for o in objects:
      key = o['Key']
      data = s3.get_object(Bucket='sd-vehicle-data', Key=key)
      dfs.append(pd.read_csv(data['Body'], delimiter=delimiter))
  return pd.concat(dfs)
  
def to_csv(df, path, boto3_session, separator=","):
  bucket = path.split("//")[1].split("/")[0]
  key = path.split("//")[1].replace(f"{bucket}/", "")
  s3.put_object(Bucket=bucket, Key=key, Body=df.to_csv(sep=separator, index=False))
  return True

s3 = {
    "to_csv": to_csv,
    "read_csv": read_csv
}
                                                  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")


# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['queryStringParameters']['async']
    if async == 'true':
        exec_type = 'Event'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.invoke(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

    

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               
  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")

# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

서버리스로 전환하기

이제 스트리밍 데이터를 인코딩/디코딩하고, 스트림 내에서 바로 데이터를 분석할 준비가 되었습니다. 여러 스트림을 사용해 일일 차량 최고 속도도 계산해 봅니다.

변환을 수행하는 Lambda

스트림 내부에서 데이터 변환하기

base64 인코딩과 디코딩

변환용 Lambda 만들기

기본 변환 람다 함수 만들기

스트림에서 데이터 분석하기

Kinesis 데이터 분석 애플리케이션 만들기

Kinesis Data Analytics vs 변환용 Lambda

Kinesis Data Analytics의 SQL 구성 요소

여러 스트림 사용하기

Kinesis Analytics에서 데이터 전달하기

일일 최고 속도 가져오기

알림을 위한 Kinesis 데이터 분석 사용하기

스트리밍 데이터 분석

마지막 장에서는 로그, 지표, 알림, 대시보드를 활용해 스트림 성능을 모니터링하는 방법을 알아봅니다. Elasticsearch를 사용하고, 직접 Kibana 대시보드를 구축합니다.

스트리밍 데이터 사례 연구

Elasticsearch 클러스터 생성하기

RedShift vs Elasticsearch

성능 모니터링

CloudWatch 대시보드와 알람

CloudWatch 구성 요소

CloudWatch 모니터링 흐름

스트리밍 데이터 시각화

Kibana로 ElasticSearch 다루기

CloudWatch vs Elasticsearch

Elasticsearch 시각화 흐름

대안 접근 방식

또 다른 대안 접근법

스트리밍 데이터 모니터링과 시각화

사기 거래를 막아 결제를 안전하게 지키거나, 교통 흐름을 원활하게 만드는 신호 체계처럼, 우리가 일상에서 무심코 사용하는 시스템은 무엇으로 작동할까요? 정답은 스트리밍 데이터(여러 소스에서 지속적으로 생성되는 데이터)와 Amazon Kinesis, AWS Lambda 같은 서버리스 기술입니다.
<br /> <br />
이 과정에서는 가상의 데이터 엔지니어 Cody를 도우면서 이러한 강력한 기술을 활용하는 방법을 배웁니다. 목표는 시 소유 차량에서 생성되는 실시간 스트리밍 데이터를 수집하고, 이를 분석해 과속 경고와 같은 알림을 운전자에게 보내는 것입니다. Amazon Kinesis와 Firehose를 사용해 수백만 개 소스에서 데이터를 수집하고, Kinesis Analytics로 스트림을 흐르는 동안 데이터를 분석하는 방법을 익히게 됩니다. 또한 AWS Lambda에서 서버리스 함수를 만들어, 수신한 데이터에 따라 조건부로 동작을 트리거하도록 구성합니다. 학습을 마치면 AWS QuickSight와 CloudWatch로 실시간 ElasticSearch 대시보드를 만드는 방법을 알게 되고, Cody의 야심 찬 프로젝트도 성공적으로 지원하게 될 것입니다.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

AWS에서 서버리스 기술로 스트리밍 데이터를 처리하는 방법을 배우세요.

AWS Kinesis와 Lambda로 배우는 스트리밍 데이터

AWS의 서버리스 기술로 스트리밍 데이터를 다루는 방법을 학습하세요.

Data Engineer

람다 레이어 추가

실습해 봅시다!

서버리스로 전환하기 - Lambda 레이어 추가하기

Lambda 레이어 추가하기

Create Your Free Account