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

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

Options

Batch

Stream

배치 vs 스트림

Data Producers

..put data into Firehose Stream

...which delivers it to Destination

프로듀서와 대상지

Firehose 딜리버리 스트림 관리

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

역할 생성하기

IAM 사용자

역할(Role)

IAM 사용자와 역할 모두

IAM 사용자 vs. 역할

Permissions for User only

Permissions for User and Role

권한 연습

S3 버킷 생성

Firehose 전송 스트림 다루기

# _setup.py: Create clients, load data, etc. No need to edit.

import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.____(
    DeliveryStreamName="gps-delivery-stream",
    DeliveryStreamType="DirectPut",
    # Specify configuration of the destination S3 bucket
    ____ = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


# _setup.py: Create clients, load data, etc. No need to edit.
import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


첫 번째 Firehose 스트림 만들기

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).head(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.____(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'____': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


Firehose 스트림에 기록하기

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).sample(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# _run_deps.py: Execute dependencies. No need to edit.
import create_firehose
import OBD2_sensors


# create_firehose.py: Create firehose stream and s3 bucket. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write OBD2 sensor data. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})
    


# analyze_data.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.____(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.____(Bucket='sd-vehicle-data', ____=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())

# OBD2_sensors.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.list_objects(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.get_object(Bucket='sd-vehicle-data', Key=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())


Firehose 데이터 읽기

클라우드에서의 스트리밍

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

서버리스로 가기

Lambda 함수 생성 및 실행

Serverless vs servers

Lambda 함수 구성 요소

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

첫 번째 라이브 Lambda!

Lambda 레이어 추가하기

Lambda 흐름

환경 변수 사용하기

서버리스 데이터 워크플로

빠른 Lambda vs 느린 Lambda

시간 트리거 Lambda

시간 트리거 Lambda 코드

서버리스 API

API Lambda 만들기

서버리스로 전환하기

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

변환을 수행하는 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

코디와 차량대

텔레매틱스 스트리밍

AWS Kinesis

Data Firehose

전송 스트림

Firehose 클라이언트 생성

전송 스트림 작업하기

스트림 삭제

복습

연습해 봅시다!

클라우드에서의 스트리밍 - 스트리밍이란 무엇이며 왜 중요할까요?

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

Create Your Free Account