AWS Data Platform Workshop

node --version
# v18.x.x 이상 확인

npm --version
# 9.x.x 이상 확인

npm install -g aws-cdk

# 설치 확인
cdk --version

aws configure
# AWS Access Key ID: [입력]
# AWS Secret Access Key: [입력]
# Default region name: ap-northeast-2
# Default output format: json

# 설정 확인
aws sts get-caller-identity

mkdir aws-data-platform-workshop
cd aws-data-platform-workshop

cdk init app --language typescript

# 필요한 패키지 설치
npm install @aws-cdk/aws-s3 @aws-cdk/aws-glue @aws-cdk/aws-kinesis \
  @aws-cdk/aws-lambda @aws-cdk/aws-iam @aws-cdk/aws-athena

cdk bootstrap aws://ACCOUNT_ID/ap-northeast-2

# ACCOUNT_ID는 본인의 AWS 계정 ID로 대체

import * as cdk from 'aws-cdk-lib';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as glue from 'aws-cdk-lib/aws-glue';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Construct } from 'constructs';

export class DataLakeStack extends cdk.Stack {
  public readonly rawBucket: s3.Bucket;
  public readonly processedBucket: s3.Bucket;
  public readonly curatedBucket: s3.Bucket;
  public readonly database: glue.CfnDatabase;

  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Raw 데이터 버킷 (원본 데이터)
    this.rawBucket = new s3.Bucket(this, 'RawDataBucket', {
      bucketName: `data-lake-raw-${this.account}-${this.region}`,
      versioned: true,
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      lifecycleRules: [
        {
          id: 'MoveToIA',
          transitions: [
            {
              storageClass: s3.StorageClass.INFREQUENT_ACCESS,
              transitionAfter: cdk.Duration.days(90),
            },
            {
              storageClass: s3.StorageClass.GLACIER,
              transitionAfter: cdk.Duration.days(180),
            },
          ],
        },
      ],
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });

    // Processed 데이터 버킷 (정제된 데이터)
    this.processedBucket = new s3.Bucket(this, 'ProcessedDataBucket', {
      bucketName: `data-lake-processed-${this.account}-${this.region}`,
      versioned: true,
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });

    // Curated 데이터 버킷 (분석용 데이터)
    this.curatedBucket = new s3.Bucket(this, 'CuratedDataBucket', {
      bucketName: `data-lake-curated-${this.account}-${this.region}`,
      versioned: true,
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });

    // Glue Database
    this.database = new glue.CfnDatabase(this, 'GlueDatabase', {
      catalogId: this.account,
      databaseInput: {
        name: 'data_platform_db',
        description: 'Data Platform Workshop Database',
        locationUri: `s3://${this.curatedBucket.bucketName}/`,
      },
    });

    // Outputs
    new cdk.CfnOutput(this, 'RawBucketName', {
      value: this.rawBucket.bucketName,
      description: 'Raw data bucket name',
    });

    new cdk.CfnOutput(this, 'ProcessedBucketName', {
      value: this.processedBucket.bucketName,
      description: 'Processed data bucket name',
    });

    new cdk.CfnOutput(this, 'CuratedBucketName', {
      value: this.curatedBucket.bucketName,
      description: 'Curated data bucket name',
    });
  }
}

#!/usr/bin/env node
import 'source-map-support/register';
import * as cdk from 'aws-cdk-lib';
import { DataLakeStack } from '../lib/data-lake-stack';

const app = new cdk.App();

const env = {
  account: process.env.CDK_DEFAULT_ACCOUNT,
  region: process.env.CDK_DEFAULT_REGION || 'ap-northeast-2',
};

new DataLakeStack(app, 'DataLakeStack', { env });

# 변경사항 확인
cdk diff DataLakeStack

# 배포
cdk deploy DataLakeStack --require-approval never

# 샘플 데이터 생성
cat > sample_orders.json << 'EOF'
{"order_id": "ORD001", "customer_id": "C001", "product": "Laptop", "amount": 1200, "order_date": "2024-01-15"}
{"order_id": "ORD002", "customer_id": "C002", "product": "Mouse", "amount": 25, "order_date": "2024-01-15"}
{"order_id": "ORD003", "customer_id": "C001", "product": "Keyboard", "amount": 75, "order_date": "2024-01-16"}
{"order_id": "ORD004", "customer_id": "C003", "product": "Monitor", "amount": 350, "order_date": "2024-01-16"}
{"order_id": "ORD005", "customer_id": "C002", "product": "Webcam", "amount": 80, "order_date": "2024-01-17"}
EOF

# S3에 업로드
aws s3 cp sample_orders.json s3://data-lake-raw-$(aws sts get-caller-identity --query Account --output text)-ap-northeast-2/orders/year=2024/month=01/

import * as cdk from 'aws-cdk-lib';
import * as glue from 'aws-cdk-lib/aws-glue';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as s3 from 'aws-cdk-lib/aws-s3';
import { Construct } from 'constructs';

interface GlueEtlStackProps extends cdk.StackProps {
  rawBucket: s3.IBucket;
  processedBucket: s3.IBucket;
  curatedBucket: s3.IBucket;
  databaseName: string;
}

export class GlueEtlStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: GlueEtlStackProps) {
    super(scope, id, props);

    // Glue IAM Role
    const glueRole = new iam.Role(this, 'GlueRole', {
      assumedBy: new iam.ServicePrincipal('glue.amazonaws.com'),
      managedPolicies: [
        iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSGlueServiceRole'),
      ],
    });

    // S3 버킷 접근 권한
    props.rawBucket.grantRead(glueRole);
    props.processedBucket.grantReadWrite(glueRole);
    props.curatedBucket.grantReadWrite(glueRole);

    // Raw 데이터 Crawler
    const rawCrawler = new glue.CfnCrawler(this, 'RawDataCrawler', {
      name: 'raw-data-crawler',
      role: glueRole.roleArn,
      databaseName: props.databaseName,
      targets: {
        s3Targets: [
          {
            path: `s3://${props.rawBucket.bucketName}/orders/`,
          },
        ],
      },
      schemaChangePolicy: {
        updateBehavior: 'UPDATE_IN_DATABASE',
        deleteBehavior: 'LOG',
      },
      configuration: JSON.stringify({
        Version: 1.0,
        Grouping: {
          TableGroupingPolicy: 'CombineCompatibleSchemas',
        },
      }),
    });

    // ETL Job Script (S3에 저장)
    const etlScript = `
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
from pyspark.sql.functions import col, to_date, year, month

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'SOURCE_PATH', 'TARGET_PATH'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read raw JSON data
df = spark.read.json(args['SOURCE_PATH'])

# Transform: Add date partitions, clean data
df_transformed = df \
    .withColumn("order_date", to_date(col("order_date"))) \
    .withColumn("year", year(col("order_date"))) \
    .withColumn("month", month(col("order_date"))) \
    .filter(col("amount") > 0)

# Write as Parquet with partitioning
df_transformed.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet(args['TARGET_PATH'])

job.commit()
`;

    // ETL Job
    const etlJob = new glue.CfnJob(this, 'OrdersEtlJob', {
      name: 'orders-etl-job',
      role: glueRole.roleArn,
      command: {
        name: 'glueetl',
        pythonVersion: '3',
        scriptLocation: `s3://${props.processedBucket.bucketName}/scripts/orders_etl.py`,
      },
      defaultArguments: {
        '--SOURCE_PATH': `s3://${props.rawBucket.bucketName}/orders/`,
        '--TARGET_PATH': `s3://${props.processedBucket.bucketName}/orders/`,
        '--job-language': 'python',
        '--enable-metrics': 'true',
        '--enable-continuous-cloudwatch-log': 'true',
      },
      glueVersion: '4.0',
      numberOfWorkers: 2,
      workerType: 'G.1X',
      timeout: 60,
    });

    // Processed 데이터 Crawler
    const processedCrawler = new glue.CfnCrawler(this, 'ProcessedDataCrawler', {
      name: 'processed-data-crawler',
      role: glueRole.roleArn,
      databaseName: props.databaseName,
      targets: {
        s3Targets: [
          {
            path: `s3://${props.processedBucket.bucketName}/orders/`,
          },
        ],
      },
      schemaChangePolicy: {
        updateBehavior: 'UPDATE_IN_DATABASE',
        deleteBehavior: 'LOG',
      },
    });

    // Outputs
    new cdk.CfnOutput(this, 'GlueRoleArn', {
      value: glueRole.roleArn,
    });
  }
}

# ETL 스크립트 생성
cat > orders_etl.py << 'EOF'
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
from pyspark.sql.functions import col, to_date, year, month

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'SOURCE_PATH', 'TARGET_PATH'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read raw JSON data
df = spark.read.json(args['SOURCE_PATH'])

# Transform
df_transformed = df \
    .withColumn("order_date", to_date(col("order_date"))) \
    .withColumn("year", year(col("order_date"))) \
    .withColumn("month", month(col("order_date"))) \
    .filter(col("amount") > 0)

# Write as Parquet
df_transformed.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet(args['TARGET_PATH'])

job.commit()
EOF

# S3에 업로드
PROCESSED_BUCKET="data-lake-processed-$(aws sts get-caller-identity --query Account --output text)-ap-northeast-2"
aws s3 cp orders_etl.py s3://$PROCESSED_BUCKET/scripts/

# 배포
cdk deploy GlueEtlStack --require-approval never

# Raw Crawler 실행
aws glue start-crawler --name raw-data-crawler

# Crawler 완료 대기 (약 1-2분)
aws glue get-crawler --name raw-data-crawler --query 'Crawler.State'

# ETL Job 실행
aws glue start-job-run --job-name orders-etl-job

# Job 상태 확인
aws glue get-job-runs --job-name orders-etl-job --query 'JobRuns[0].JobRunState'

# Processed Crawler 실행
aws glue start-crawler --name processed-data-crawler

import * as cdk from 'aws-cdk-lib';
import * as athena from 'aws-cdk-lib/aws-athena';
import * as s3 from 'aws-cdk-lib/aws-s3';
import { Construct } from 'constructs';

interface AthenaStackProps extends cdk.StackProps {
  curatedBucket: s3.IBucket;
}

export class AthenaStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: AthenaStackProps) {
    super(scope, id, props);

    // Athena 쿼리 결과 버킷
    const queryResultsBucket = new s3.Bucket(this, 'AthenaQueryResults', {
      bucketName: `athena-results-${this.account}-${this.region}`,
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      lifecycleRules: [
        {
          expiration: cdk.Duration.days(30),
        },
      ],
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      autoDeleteObjects: true,
    });

    // Athena Workgroup
    const workgroup = new athena.CfnWorkGroup(this, 'DataPlatformWorkgroup', {
      name: 'data-platform-workgroup',
      description: 'Workgroup for Data Platform Workshop',
      state: 'ENABLED',
      workGroupConfiguration: {
        resultConfiguration: {
          outputLocation: `s3://${queryResultsBucket.bucketName}/results/`,
          encryptionConfiguration: {
            encryptionOption: 'SSE_S3',
          },
        },
        enforceWorkGroupConfiguration: true,
        publishCloudWatchMetricsEnabled: true,
        bytesScannedCutoffPerQuery: 10737418240, // 10GB limit
        engineVersion: {
          selectedEngineVersion: 'Athena engine version 3',
        },
      },
    });

    new cdk.CfnOutput(this, 'WorkgroupName', {
      value: workgroup.name,
    });

    new cdk.CfnOutput(this, 'QueryResultsBucket', {
      value: queryResultsBucket.bucketName,
    });
  }
}

# Athena 콘솔에서 실행하거나 CLI 사용

# 1. 테이블 확인
aws athena start-query-execution \
  --query-string "SHOW TABLES IN data_platform_db" \
  --work-group data-platform-workgroup \
  --query-execution-context Database=data_platform_db

# 2. 데이터 조회 (파티션 활용)
aws athena start-query-execution \
  --query-string "SELECT * FROM orders WHERE year = 2024 AND month = 1 LIMIT 10" \
  --work-group data-platform-workgroup \
  --query-execution-context Database=data_platform_db

# 3. 집계 쿼리
aws athena start-query-execution \
  --query-string "SELECT product, SUM(amount) as total_sales, COUNT(*) as order_count FROM orders GROUP BY product ORDER BY total_sales DESC" \
  --work-group data-platform-workgroup \
  --query-execution-context Database=data_platform_db

-- Athena 콘솔에서 실행
CREATE TABLE data_platform_db.orders_optimized
WITH (
  format = 'PARQUET',
  parquet_compression = 'SNAPPY',
  partitioned_by = ARRAY['year', 'month'],
  external_location = 's3://data-lake-curated-ACCOUNT_ID-ap-northeast-2/orders_optimized/'
) AS
SELECT 
  order_id,
  customer_id,
  product,
  amount,
  order_date,
  year,
  month
FROM data_platform_db.orders;

import * as cdk from 'aws-cdk-lib';
import * as kinesis from 'aws-cdk-lib/aws-kinesis';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as logs from 'aws-cdk-lib/aws-logs';
import { Construct } from 'constructs';

interface StreamingStackProps extends cdk.StackProps {
  rawBucket: s3.IBucket;
}

export class StreamingStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: StreamingStackProps) {
    super(scope, id, props);

    // Kinesis Data Stream
    const stream = new kinesis.Stream(this, 'EventStream', {
      streamName: 'event-stream',
      shardCount: 1,
      retentionPeriod: cdk.Duration.hours(24),
    });

    // Lambda for real-time processing
    const processorLambda = new lambda.Function(this, 'StreamProcessor', {
      functionName: 'stream-processor',
      runtime: lambda.Runtime.PYTHON_3_11,
      handler: 'index.handler',
      code: lambda.Code.fromInline(`
import json
import base64

def handler(event, context):
    output = []
    for record in event['records']:
        payload = base64.b64decode(record['data']).decode('utf-8')
        data = json.loads(payload)
        
        # Add processing timestamp
        data['processed_at'] = context.aws_request_id
        
        # Enrich data
        if data.get('amount', 0) > 100:
            data['category'] = 'high_value'
        else:
            data['category'] = 'standard'
        
        output_record = {
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(json.dumps(data).encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)
    
    return {'records': output}
`),
      timeout: cdk.Duration.minutes(1),
      memorySize: 256,
    });

    // Firehose IAM Role
    const firehoseRole = new iam.Role(this, 'FirehoseRole', {
      assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
    });

    props.rawBucket.grantReadWrite(firehoseRole);
    processorLambda.grantInvoke(firehoseRole);
    stream.grantRead(firehoseRole);

    // CloudWatch Log Group for Firehose
    const logGroup = new logs.LogGroup(this, 'FirehoseLogGroup', {
      logGroupName: '/aws/kinesisfirehose/event-delivery',
      retention: logs.RetentionDays.ONE_WEEK,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    const logStream = new logs.LogStream(this, 'FirehoseLogStream', {
      logGroup: logGroup,
      logStreamName: 'delivery-stream',
    });

    // Kinesis Data Firehose
    const deliveryStream = new firehose.CfnDeliveryStream(this, 'EventDeliveryStream', {
      deliveryStreamName: 'event-delivery-stream',
      deliveryStreamType: 'KinesisStreamAsSource',
      kinesisStreamSourceConfiguration: {
        kinesisStreamArn: stream.streamArn,
        roleArn: firehoseRole.roleArn,
      },
      extendedS3DestinationConfiguration: {
        bucketArn: props.rawBucket.bucketArn,
        roleArn: firehoseRole.roleArn,
        prefix: 'streaming/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/',
        errorOutputPrefix: 'streaming-errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/',
        bufferingHints: {
          intervalInSeconds: 60,
          sizeInMBs: 1,
        },
        compressionFormat: 'GZIP',
        processingConfiguration: {
          enabled: true,
          processors: [
            {
              type: 'Lambda',
              parameters: [
                {
                  parameterName: 'LambdaArn',
                  parameterValue: processorLambda.functionArn,
                },
              ],
            },
          ],
        },
        cloudWatchLoggingOptions: {
          enabled: true,
          logGroupName: logGroup.logGroupName,
          logStreamName: logStream.logStreamName,
        },
      },
    });

    // Outputs
    new cdk.CfnOutput(this, 'StreamName', {
      value: stream.streamName,
    });

    new cdk.CfnOutput(this, 'DeliveryStreamName', {
      value: deliveryStream.deliveryStreamName!,
    });
  }
}

# 테스트 데이터 전송 스크립트
cat > send_events.py << 'EOF'
import boto3
import json
import time
import random
from datetime import datetime

kinesis = boto3.client('kinesis', region_name='ap-northeast-2')
stream_name = 'event-stream'

products = ['Laptop', 'Mouse', 'Keyboard', 'Monitor', 'Webcam', 'Headset']
customers = ['C001', 'C002', 'C003', 'C004', 'C005']

for i in range(100):
    event = {
        'event_id': f'EVT{i:05d}',
        'customer_id': random.choice(customers),
        'product': random.choice(products),
        'amount': random.randint(10, 500),
        'timestamp': datetime.utcnow().isoformat()
    }
    
    response = kinesis.put_record(
        StreamName=stream_name,
        Data=json.dumps(event),
        PartitionKey=event['customer_id']
    )
    
    print(f"Sent: {event['event_id']} - Shard: {response['ShardId']}")
    time.sleep(0.1)

print("Done!")
EOF

# 실행
pip install boto3
python send_events.py

# QuickSight는 콘솔에서 설정합니다

# 1. AWS 콘솔 → QuickSight 이동
# 2. Sign up for QuickSight (Enterprise Edition 권장)
# 3. QuickSight가 S3, Athena에 접근할 수 있도록 권한 설정

# CLI로 QuickSight 데이터 소스 생성
aws quicksight create-data-source \
  --aws-account-id $(aws sts get-caller-identity --query Account --output text) \
  --data-source-id athena-data-source \
  --name "Athena Data Source" \
  --type ATHENA \
  --data-source-parameters '{"AthenaParameters":{"WorkGroup":"data-platform-workgroup"}}' \
  --permissions '[{"Principal":"arn:aws:quicksight:ap-northeast-2:ACCOUNT_ID:user/default/ADMIN_USER","Actions":["quicksight:DescribeDataSource","quicksight:DescribeDataSourcePermissions","quicksight:PassDataSource","quicksight:UpdateDataSource","quicksight:DeleteDataSource","quicksight:UpdateDataSourcePermissions"]}]'

# QuickSight 콘솔에서:
# 1. Datasets → New dataset
# 2. Athena 선택
# 3. data_platform_db 데이터베이스 선택
# 4. orders 테이블 선택
# 5. Import to SPICE 선택 (빠른 응답을 위해)

# 또는 Custom SQL 사용:
SELECT 
  order_id,
  customer_id,
  product,
  amount,
  order_date,
  CASE 
    WHEN amount > 200 THEN 'High'
    WHEN amount > 50 THEN 'Medium'
    ELSE 'Low'
  END as value_segment
FROM data_platform_db.orders
WHERE year = 2024

# QuickSight 콘솔에서 Analysis 생성:

# 1. 매출 추이 (Line Chart)
#    - X축: order_date
#    - Y축: SUM(amount)

# 2. 제품별 매출 (Bar Chart)
#    - X축: product
#    - Y축: SUM(amount)
#    - 정렬: 내림차순

# 3. 고객별 주문 수 (Pie Chart)
#    - Group by: customer_id
#    - Value: COUNT(order_id)

# 4. KPI 카드
#    - 총 매출: SUM(amount)
#    - 총 주문 수: COUNT(order_id)
#    - 평균 주문 금액: AVG(amount)

# 5. 필터 추가
#    - order_date 범위 필터
#    - product 다중 선택 필터

# QuickSight 콘솔에서:
# 1. Dashboards → 삭제
# 2. Analyses → 삭제
# 3. Datasets → 삭제
# 4. Data sources → 삭제

# 또는 QuickSight 구독 취소 (모든 리소스 삭제됨)
# QuickSight → Manage QuickSight → Account settings → Delete account

# 모든 스택 삭제 (역순으로)
cdk destroy StreamingStack --force
cdk destroy AthenaStack --force
cdk destroy GlueEtlStack --force
cdk destroy DataLakeStack --force

# 또는 한번에 모두 삭제
cdk destroy --all --force

# CloudFormation 스택 확인
aws cloudformation list-stacks --stack-status-filter CREATE_COMPLETE UPDATE_COMPLETE

# S3 버킷 확인
aws s3 ls | grep data-lake

# Glue 리소스 확인
aws glue get-databases
aws glue get-crawlers
aws glue get-jobs

# Kinesis 확인
aws kinesis list-streams
aws firehose list-delivery-streams