💡 프로 팁: 처음부터 너무 복잡한 아키텍처를 설계하지 마. 먼저 핵심 요구사항을 충족하는 MVP(최소 기능 제품)를 설계하고, 점진적으로 확장해 나가는 것이 좋아!

🔧 구현 예시: AWS 환경에서 Terraform을 사용한 데이터 레이크하우스 인프라 설정

# S3 버킷 생성
resource "aws_s3_bucket" "data_lakehouse" {
  bucket = "my-data-lakehouse"
  acl    = "private"
  
  versioning {
    enabled = true
  }
  
  server_side_encryption_configuration {
    rule {
      apply_server_side_encryption_by_default {
        sse_algorithm = "AES256"
      }
    }
  }
  
  lifecycle_rule {
    enabled = true
    
    transition {
      days          = 90
      storage_class = "INTELLIGENT_TIERING"
    }
  }
}

# Glue 데이터 카탈로그 데이터베이스 생성
resource "aws_glue_catalog_database" "lakehouse_db" {
  name = "lakehouse_db"
}

# EMR 클러스터 설정
resource "aws_emr_cluster" "lakehouse_cluster" {
  name          = "lakehouse-cluster"
  release_label = "emr-6.6.0"
  applications  = ["Spark", "Hive", "Presto"]
  
  ec2_attributes {
    subnet_id                         = aws_subnet.main.id
    emr_managed_master_security_group = aws_security_group.master.id
    emr_managed_slave_security_group  = aws_security_group.slave.id
    instance_profile                  = aws_iam_instance_profile.emr_profile.arn
  }
  
  master_instance_group {
    instance_type = "m5.xlarge"
  }
  
  core_instance_group {
    instance_type  = "m5.xlarge"
    instance_count = 2
    
    ebs_config {
      size                 = "100"
      type                 = "gp2"
      volumes_per_instance = 1
    }
  }
  
  configurations_json = <<EOF
  [
    {
      "Classification": "spark-defaults",
      "Properties": {
        "spark.dynamicAllocation.enabled": "true",
        "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
        "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog"
      }
    }
  ]
  EOF
}
        

📊 구현 예시: Apache Airflow를 사용한 데이터 수집 파이프라인

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.providers.amazon.aws.operators.glue import AwsGlueJobOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'sales_data_pipeline',
    default_args=default_args,
    description='Load sales data into data lakehouse',
    schedule_interval='0 2 * * *',
    catchup=False
)

# 데이터 소스 파일 확인
check_source_file = S3KeySensor(
    task_id='check_source_file',
    bucket_key='raw-data/sales/{{ ds }}/sales_data.csv',
    bucket_name='source-data-bucket',
    aws_conn_id='aws_default',
    timeout=60 * 60,
    poke_interval=60,
    dag=dag
)

# 데이터 추출 및 변환 작업
extract_transform = AwsGlueJobOperator(
    task_id='extract_transform_sales_data',
    job_name='sales_data_etl',
    script_location='s3://glue-scripts/sales_etl.py',
    script_args={
        '--source_path': 's3://source-data-bucket/raw-data/sales/{{ ds }}/sales_data.csv',
        '--target_path': 's3://my-data-lakehouse/bronze/sales/date={{ ds }}/'
    },
    aws_conn_id='aws_default',
    dag=dag
)

# 데이터 품질 검사
def run_data_quality_checks(**kwargs):
    # 데이터 품질 검사 로직 구현
    pass

data_quality = PythonOperator(
    task_id='run_data_quality_checks',
    python_callable=run_data_quality_checks,
    provide_context=True,
    dag=dag
)

# Silver 레이어로 데이터 처리
process_to_silver = AwsGlueJobOperator(
    task_id='process_to_silver',
    job_name='sales_bronze_to_silver',
    script_location='s3://glue-scripts/bronze_to_silver.py',
    script_args={
        '--source_path': 's3://my-data-lakehouse/bronze/sales/date={{ ds }}/',
        '--target_path': 's3://my-data-lakehouse/silver/sales/'
    },
    aws_conn_id='aws_default',
    dag=dag
)

# Gold 레이어로 데이터 처리
process_to_gold = AwsGlueJobOperator(
    task_id='process_to_gold',
    job_name='sales_silver_to_gold',
    script_location='s3://glue-scripts/silver_to_gold.py',
    script_args={
        '--source_path': 's3://my-data-lakehouse/silver/sales/',
        '--target_path': 's3://my-data-lakehouse/gold/sales_aggregates/'
    },
    aws_conn_id='aws_default',
    dag=dag
)

# 워크플로우 정의
check_source_file >> extract_transform >> data_quality >> process_to_silver >> process_to_gold
        

🔐 구현 예시: AWS Lake Formation을 사용한 데이터 레이크하우스 보안 설정

# AWS CLI를 사용한 Lake Formation 권한 설정 예시

# 데이터베이스 생성
aws glue create-database --database-input '{"Name":"sales_db", "Description":"Sales data database"}'

# Lake Formation에 리소스 등록
aws lakeformation register-resource --resource-arn arn:aws:s3:::my-data-lakehouse --use-service-linked-role

# 데이터 위치 권한 부여
aws lakeformation grant-permissions --principal DataLakeAdmin \
  --permissions "DATA_LOCATION_ACCESS" \
  --resource '{"DataLocation": {"ResourceArn": "arn:aws:s3:::my-data-lakehouse"}}'

# 테이블 생성 권한 부여
aws lakeformation grant-permissions --principal DataEngineer \
  --permissions "CREATE_TABLE", "ALTER", "DROP" \
  --resource '{"Database": {"Name": "sales_db"}}'

# 테이블 데이터 접근 권한 부여
aws lakeformation grant-permissions --principal DataAnalyst \
  --permissions "SELECT" \
  --resource '{"Table": {"DatabaseName": "sales_db", "Name": "sales_silver"}}'

# 열 수준 권한 부여 (민감한 정보 제한)
aws lakeformation grant-permissions --principal DataAnalyst \
  --permissions "SELECT" \
  --resource '{"TableWithColumns": {"DatabaseName": "sales_db", "Name": "customer_data", "ColumnNames": ["customer_id", "purchase_date", "product_id", "quantity"]}}'
        

🔮 구현 예시: Databricks를 사용한 머신러닝 파이프라인

# PySpark을 사용한 머신러닝 파이프라인 예시

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
import mlflow
import mlflow.spark

# Spark 세션 생성
spark = SparkSession.builder \
    .appName("Customer Churn Prediction") \
    .getOrCreate()

# 데이터 로드
customer_data = spark.read.format("delta").load("/data/gold/customer_features")

# 특성 준비
feature_cols = ["age", "tenure", "monthly_charges", "total_charges", "num_products"]
assembler = VectorAssembler(inputCols=feature_cols, outputCol="features_raw")
scaler = StandardScaler(inputCol="features_raw", outputCol="features")

# 모델 정의
rf = RandomForestClassifier(labelCol="churn", featuresCol="features", numTrees=100)

# 파이프라인 구성
pipeline = Pipeline(stages=[assembler, scaler, rf])

# 하이퍼파라미터 튜닝 설정
paramGrid = ParamGridBuilder() \
    .addGrid(rf.maxDepth, [5, 10, 15]) \
    .addGrid(rf.minInstancesPerNode, [1, 5, 10]) \
    .build()

# 교차 검증 설정
crossval = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=paramGrid,
    evaluator=BinaryClassificationEvaluator(labelCol="churn"),
    numFolds=3
)

# MLflow 실험 추적 설정
mlflow.set_experiment("/Shared/customer_churn_prediction")

with mlflow.start_run(run_name="rf_model_training") as run:
    # 모델 훈련
    train_data, test_data = customer_data.randomSplit([0.8, 0.2], seed=42)
    model = crossval.fit(train_data)
    
    # 모델 평가
    predictions = model.transform(test_data)
    evaluator = BinaryClassificationEvaluator(labelCol="churn")
    auc = evaluator.evaluate(predictions)
    
    # 매개변수 및 메트릭 로깅
    best_model = model.bestModel.stages[-1]
    mlflow.log_param("numTrees", best_model.getNumTrees)
    mlflow.log_param("maxDepth", best_model.getMaxDepth())
    mlflow.log_metric("auc", auc)
    
    # 모델 저장
    mlflow.spark.log_model(model.bestModel, "spark-model")
    
    # 모델 등록
    model_uri = f"runs:/{run.info.run_id}/spark-model"
    mlflow.register_model(model_uri, "customer_churn_predictor")
    
# 모델 배포 (예: REST API로 서빙)
# 이 부분은 MLflow Model Serving 또는 다른 서빙 플랫폼을 사용하여 구현
        

📈 구현 예시: Grafana와 Prometheus를 사용한 모니터링 대시보드

# Prometheus 구성 예시 (prometheus.yml)

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'spark_metrics'
    static_configs:
      - targets: ['spark-master:7777']
  
  - job_name: 'lakehouse_api'
    static_configs:
      - targets: ['api-gateway:8080']
  
  - job_name: 'airflow'
    static_configs:
      - targets: ['airflow-webserver:8080']

# Grafana 대시보드 JSON 스니펫
{
  "dashboard": {
    "id": null,
    "title": "Data Lakehouse Monitoring",
    "panels": [
      {
        "title": "Query Execution Time",
        "type": "graph",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "spark_executor_query_time_seconds",
            "legendFormat": "{{query_type}}"
          }
        ]
      },
      {
        "title": "Data Pipeline Status",
        "type": "stat",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "sum(airflow_dag_status{status=\"success\"}) / sum(airflow_dag_status) * 100",
            "legendFormat": "Success Rate"
          }
        ]
      },
      {
        "title": "Storage Usage",
        "type": "gauge",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "sum(lakehouse_storage_bytes) / 1024 / 1024 / 1024",
            "legendFormat": "Total GB"
          }
        ]
      }
    ]
  }
}