[AWS] 데이터 & 분석

Amazon Athena

[S3 Bucket] ← CSV, JSON, Parquet 파일
    ↓
[Athena] ← SQL 쿼리
    ↓
결과 반환 (수 초)

• 서버리스 (인프라 관리 불필요)

• 표준 SQL 지원

• 스캔한 데이터만큼만 과금 ($5/TB)

• CSV, JSON, Parquet, ORC 지원

사용 방법

-- 1. 테이블 생성 (S3 데이터 연결)
CREATE EXTERNAL TABLE logs (
    timestamp STRING,
    level STRING,
    message STRING,
    user_id STRING
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES ('field.delim' = ',')
LOCATION 's3://my-logs-bucket/logs/';

-- 2. 쿼리 실행
SELECT
    level,
    COUNT(*) as count,
    DATE(timestamp) as date
FROM logs
WHERE timestamp > '2024-11-01'
GROUP BY level, DATE(timestamp)
ORDER BY date DESC;

-- 3. 파티션 활용 (비용 절감)
CREATE EXTERNAL TABLE logs_partitioned (
    timestamp STRING,
    level STRING,
    message STRING
)
PARTITIONED BY (year STRING, month STRING, day STRING)
STORED AS PARQUET
LOCATION 's3://my-logs-bucket/partitioned/';

-- 파티션별 쿼리 (특정 날짜만 스캔)
SELECT * FROM logs_partitioned
WHERE year='2024' AND month='11' AND day='19';

성능 최적화

1. 파티셔닝

❌ 비효율적:
s3://bucket/logs/log1.csv (1TB 전체 스캔)
비용: $5

✅ 효율적:
s3://bucket/logs/year=2024/month=11/day=19/log.csv
비용: $0.01 (필요한 날짜만)

2. 컬럼 형식 (Parquet/ORC)

CSV: 1GB 스캔 = $5/TB
Parquet: 100MB 스캔 = $0.50/TB (10배 절감 + 빠름)

# CSV → Parquet 변환
import pandas as pd

df = pd.read_csv('data.csv')
df.to_parquet('data.parquet', compression='snappy')

3. 압축

압축 없음: 10GB
Snappy 압축: 2GB (5배 감소)
→ 비용 5배 절감

실전 활용

# Python에서 Athena 쿼리
import boto3
import time

athena = boto3.client('athena')

def query_athena(query):
    # 쿼리 실행
    response = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={'Database': 'my_database'},
        ResultConfiguration={'OutputLocation': 's3://my-results/'}
    )

    query_id = response['QueryExecutionId']

    # 완료 대기
    while True:
        status = athena.get_query_execution(QueryExecutionId=query_id)
        state = status['QueryExecution']['Status']['State']

        if state in ['SUCCEEDED', 'FAILED', 'CANCELLED']:
            break
        time.sleep(1)

    # 결과 가져오기
    if state == 'SUCCEEDED':
        results = athena.get_query_results(QueryExecutionId=query_id)
        return results['ResultSet']['Rows']

# 사용
results = query_athena("SELECT * FROM logs WHERE level='ERROR' LIMIT 10")

Amazon Redshift

[S3] → [Redshift] → 복잡한 분석 쿼리
[RDS] ↗              초고속 응답

• 컬럼 기반 스토리지

• 대규모 조인 및 집계 최적화

• Redshift Spectrum (S3 쿼리)

• 노드 확장 가능

사용 방법

-- 테이블 생성
CREATE TABLE sales (
    sale_id INT,
    product_id INT,
    user_id INT,
    amount DECIMAL(10,2),
    sale_date DATE
)
DISTKEY(user_id)
SORTKEY(sale_date);

-- S3에서 데이터 로드 (COPY)
COPY sales
FROM 's3://my-bucket/sales-data/'
IAM_ROLE 'arn:aws:iam::account:role/RedshiftRole'
FORMAT AS PARQUET;

-- 복잡한 분석 쿼리
SELECT
    DATE_TRUNC('month', sale_date) as month,
    product_id,
    SUM(amount) as total_sales,
    COUNT(DISTINCT user_id) as unique_customers
FROM sales
WHERE sale_date >= '2024-01-01'
GROUP BY month, product_id
ORDER BY month, total_sales DESC;

Redshift Spectrum

-- External 테이블 생성
CREATE EXTERNAL SCHEMA s3_data
FROM DATA CATALOG
DATABASE 's3_database'
IAM_ROLE 'arn:aws:iam::account:role/RedshiftRole';

-- S3 + Redshift 조인
SELECT
    r.user_id,
    r.total_purchases,
    s.total_clicks
FROM redshift_table r
JOIN s3_data.clickstream s ON r.user_id = s.user_id;

OpenSearch (구 ElasticSearch)

[Application] → [OpenSearch]
                    ↓
                검색, 필터링, 집계
                    ↓
                [Kibana] ← 시각화

• 전문 검색 (풀텍스트)

• 로그 분석

• 실시간 분석

• 보안 모니터링

사용 방법

from opensearchpy import OpenSearch

client = OpenSearch(
    hosts=[{'host': 'search-domain.region.es.amazonaws.com', 'port': 443}],
    http_auth=('username', 'password'),
    use_ssl=True
)

# 문서 인덱싱
client.index(
    index='products',
    id='1',
    body={
        'title': 'iPhone 15 Pro',
        'description': 'Latest Apple smartphone',
        'price': 999,
        'category': 'Electronics'
    }
)

# 검색
response = client.search(
    index='products',
    body={
        'query': {
            'match': {
                'description': 'smartphone'
            }
        },
        'aggs': {
            'avg_price': {
                'avg': {'field': 'price'}
            }
        }
    }
)

EMR (Elastic MapReduce)

[EMR Cluster]
├─ Master Node (조정)
├─ Core Nodes (데이터 저장 + 처리)
└─ Task Nodes (처리만)

• 대규모 데이터 처리 (ETL)

• 머신러닝 학습

• 로그 분석

• 데이터 과학

Spark 예시

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DataAnalysis").getOrCreate()

# S3에서 데이터 읽기
df = spark.read.parquet("s3://my-bucket/data/")

# 데이터 처리
result = df.filter(df['amount'] > 100) \
           .groupBy('category') \
           .sum('amount') \
           .orderBy('sum(amount)', ascending=False)

# 결과 저장
result.write.parquet("s3://my-bucket/results/")

QuickSight

데이터 소스:
[S3, RDS, Redshift, Athena] → [QuickSight] → 대시보드

• 서버리스

• ML 인사이트

• 임베딩 가능

• 모바일 앱

사용 예시

대시보드 구성:
1. 데이터셋 연결 (Athena/Redshift)
2. 시각화 생성
   - 선 그래프: 시간별 매출
   - 막대 그래프: 제품별 판매량
   - 파이 차트: 카테고리 비율
3. 필터 추가 (날짜, 지역 등)
4. 공유 또는 임베딩

AWS Glue

[데이터 소스] → [Glue Crawler] → [Data Catalog]
                                       ↓
                                  [Glue ETL Job]
                                       ↓
                                  [대상 저장소]

1. Glue Crawler: 데이터 스키마 자동 발견

2. Data Catalog: 메타데이터 저장소

3. Glue ETL: 데이터 변환 작업

실습

1. Crawler로 스키마 발견

# Crawler 생성
aws glue create-crawler \
  --name my-s3-crawler \
  --role arn:aws:iam::account:role/GlueRole \
  --database-name my_database \
  --targets '{
    "S3Targets": [{
      "Path": "s3://my-bucket/data/"
    }]
  }'

# Crawler 실행
aws glue start-crawler --name my-s3-crawler

2. ETL Job 작성

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

# Glue Job 초기화
glueContext = GlueContext(SparkContext.getOrCreate())
spark = glueContext.spark_session
job = Job(glueContext)

# 데이터 읽기
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="my_database",
    table_name="raw_data"
)

# 변환
# 1. 불필요한 컬럼 제거
transformed = DropFields.apply(
    frame=datasource,
    paths=["temp_column"]
)

# 2. 데이터 타입 변경
transformed = ApplyMapping.apply(
    frame=transformed,
    mappings=[
        ("id", "string", "id", "int"),
        ("timestamp", "string", "timestamp", "timestamp"),
        ("amount", "string", "amount", "decimal(10,2)")
    ]
)

# 3. 필터링
transformed = Filter.apply(
    frame=transformed,
    f=lambda x: x["amount"] > 0
)

# 결과 저장 (Parquet)
glueContext.write_dynamic_frame.from_options(
    frame=transformed,
    connection_type="s3",
    connection_options={"path": "s3://my-bucket/processed/"},
    format="parquet"
)

job.commit()

Lake Formation

[Lake Formation]
├─ 데이터 수집 (S3, RDS, etc.)
├─ 데이터 카탈로그 (Glue)
├─ 보안 및 접근 제어
└─ 데이터 공유
    ↓
[Athena, Redshift, EMR] ← 분석

• 중앙화된 권한 관리

• 컬럼/행 수준 보안

• 데이터 공유 (Cross-Account)

Kinesis Data Analytics

[Kinesis Data Streams]
    ↓
[Kinesis Data Analytics] ← SQL 쿼리
    ↓
[결과 스트림] → Lambda, S3, Redshift

예시

-- 실시간 이동 평균
CREATE OR REPLACE STREAM "DESTINATION_STREAM" (
    stock_symbol VARCHAR(4),
    avg_price DOUBLE
);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS
INSERT INTO "DESTINATION_STREAM"
SELECT STREAM
    stock_symbol,
    AVG(price) OVER (
        PARTITION BY stock_symbol
        RANGE INTERVAL '1' MINUTE PRECEDING
    ) as avg_price
FROM "SOURCE_STREAM"
WHERE price > 0;

MSK (Managed Streaming for Apache Kafka)

[Producer] → [MSK Cluster] → [Consumer]
                 ↓
           [Topic: orders]
           [Topic: events]

빅데이터 파이프라인 예시

시나리오: 실시간 전자상거래 분석

[웹사이트/앱] (클릭, 주문)
    ↓
[Kinesis Data Streams]
    ├─→ [Kinesis Data Analytics] → 실시간 대시보드
    ├─→ [Lambda] → [DynamoDB] ← 실시간 추천
    └─→ [Kinesis Firehose]
            ↓
        [S3: Raw Data]
            ↓
        [Glue Crawler] → [Data Catalog]
            ↓
        [Glue ETL Job]
            ├─ 데이터 정제
            ├─ 파티셔닝
            └─ Parquet 변환
            ↓
        [S3: Processed Data]
            ↓
        [Athena] ← 임시 분석
        [Redshift] ← 정기 리포트
        [QuickSight] ← 대시보드

비용 최적화

Athena

1. Parquet/ORC 형식 사용 (10배 절감)
2. 파티셔닝 (필요한 데이터만 스캔)
3. 컬럼 선택 (SELECT *)
4. 압축 (Snappy, Gzip)

예시:
CSV 10TB 스캔: $50
Parquet 1TB 스캔: $5 (90% 절감)

Redshift

1. Reserved Instances (75% 할인)
2. Pause/Resume (사용하지 않을 때)
3. Concurrency Scaling (burst만)
4. Spectrum (자주 쓰지 않는 데이터는 S3)

EMR

1. Spot Instances (70-90% 할인)
2. Task 노드만 Spot 사용
3. S3 기반 (EMRFS) - 클러스터 종료 가능
4. Auto Scaling