데이터 레이크

# AWS S3 기반 데이터 레이크 아키텍처
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import boto3

# S3 데이터 레이크 계층 구조
DATA_LAKE = "s3://company-datalake"
ZONES = {
    "raw": f"{DATA_LAKE}/raw/",      # 원시 데이터
    "bronze": f"{DATA_LAKE}/bronze/", # 정제된 데이터
    "silver": f"{DATA_LAKE}/silver/", # 변환/집계 데이터
    "gold": f"{DATA_LAKE}/gold/"      # 분석용 마트
}

# Spark 세션 초기화
spark = SparkSession.builder \
    .appName("DataLakeProcessing") \
    .config("spark.sql.extensions",
            "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# Raw Zone: 원시 JSON 데이터 수집
raw_events = spark.read \
    .option("multiLine", True) \
    .json(f"{ZONES['raw']}/events/2024/01/15/")

# Bronze Zone: 정제 및 파티셔닝
bronze_events = raw_events \
    .dropDuplicates(["event_id"]) \
    .withColumn("event_date", to_date("timestamp")) \
    .withColumn("event_hour", hour("timestamp"))

bronze_events.write \
    .partitionBy("event_date", "event_hour") \
    .mode("append") \
    .parquet(f"{ZONES['bronze']}/events/")

print(f"Processed {bronze_events.count()} events to Bronze Zone")

# AWS Glue Catalog에 테이블 등록
glue = boto3.client('glue')
glue.create_table(
    DatabaseName='bronze_db',
    TableInput={
        'Name': 'events',
        'StorageDescriptor': {
            'Location': f"{ZONES['bronze']}/events/",
            'InputFormat': 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
        }
    }
)