Session 9: DataOps & MLOps

┌─────────────────────────────────────────────────────────────────────┐
│                    DataOps Pipeline                                  │
│                                                                      │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐      │
│  │  Code    │───▶│  Build   │───▶│  Test    │───▶│  Deploy  │      │
│  │          │    │          │    │          │    │          │      │
│  │ • ETL    │    │ • Package│    │ • Unit   │    │ • Dev    │      │
│  │ • Schema │    │ • Docker │    │ • Data   │    │ • Stage  │      │
│  │ • Config │    │ • Glue   │    │ • Quality│    │ • Prod   │      │
│  └──────────┘    └──────────┘    └──────────┘    └──────────┘      │
│       │                                               │              │
│       │              ┌──────────┐                     │              │
│       └──────────────│ Monitor  │◀────────────────────┘              │
│                      │          │                                    │
│                      │ • Metrics│                                    │
│                      │ • Alerts │                                    │
│                      │ • Logs   │                                    │
│                      └──────────┘                                    │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    MLOps Pipeline                                    │
│                                                                      │
│  Data Preparation          Model Development         Deployment      │
│  ┌─────────────┐          ┌─────────────┐          ┌─────────────┐  │
│  │ Feature     │          │ Training    │          │ Endpoint    │  │
│  │ Engineering │─────────▶│ & Tuning    │─────────▶│ Deployment  │  │
│  │             │          │             │          │             │  │
│  │ • S3        │          │ • SageMaker │          │ • Real-time │  │
│  │ • Glue      │          │ • HPO       │          │ • Batch     │  │
│  │ • Feature   │          │ • Experiments│         │ • Serverless│  │
│  │   Store     │          │             │          │             │  │
│  └─────────────┘          └─────────────┘          └─────────────┘  │
│        │                        │                        │          │
│        │                        │                        │          │
│        ▼                        ▼                        ▼          │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │                    Model Registry                                ││
│  │  • Version Control  • Approval Workflow  • Lineage Tracking     ││
│  └─────────────────────────────────────────────────────────────────┘│
│                              │                                       │
│                              ▼                                       │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │                    Model Monitoring                              ││
│  │  • Data Drift  • Model Drift  • Bias Detection                  ││
│  └─────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    Data Pipeline Observability                       │
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  Glue ETL Metrics                                                ││
│  │  ├── glue.driver.aggregate.numCompletedTasks                    ││
│  │  ├── glue.driver.aggregate.numFailedTasks                       ││
│  │  ├── glue.driver.ExecutorAllocationManager.executors.numberAll  ││
│  │  └── glue.driver.jvm.heap.usage                                 ││
│  └─────────────────────────────────────────────────────────────────┘│
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  Kinesis Metrics                                                 ││
│  │  ├── IncomingRecords / IncomingBytes                            ││
│  │  ├── GetRecords.IteratorAgeMilliseconds (Consumer Lag)          ││
│  │  ├── WriteProvisionedThroughputExceeded                         ││
│  │  └── ReadProvisionedThroughputExceeded                          ││
│  └─────────────────────────────────────────────────────────────────┘│
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  Redshift Metrics                                                ││
│  │  ├── CPUUtilization / PercentageDiskSpaceUsed                   ││
│  │  ├── DatabaseConnections                                         ││
│  │  ├── QueryDuration / QueriesCompletedPerSecond                  ││
│  │  └── WLMQueueLength                                              ││
│  └─────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    Model Monitoring Types                            │
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  1. Data Quality Monitoring                                      ││
│  │     • 입력 데이터의 통계적 특성 변화 감지                         ││
│  │     • Baseline vs Current 비교                                   ││
│  │     • Missing values, data types, distribution                  ││
│  └─────────────────────────────────────────────────────────────────┘│
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  2. Model Quality Monitoring                                     ││
│  │     • 예측 정확도 추적 (Ground Truth 필요)                       ││
│  │     • Accuracy, Precision, Recall, F1                           ││
│  │     • Regression: MSE, MAE, R²                                  ││
│  └─────────────────────────────────────────────────────────────────┘│
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  3. Bias Drift Monitoring                                        ││
│  │     • 모델 편향 변화 감지                                        ││
│  │     • Demographic parity, Equalized odds                        ││
│  └─────────────────────────────────────────────────────────────────┘│
│                                                                      │
│  ┌─────────────────────────────────────────────────────────────────┐│
│  │  4. Feature Attribution Drift                                    ││
│  │     • Feature importance 변화 감지                               ││
│  │     • SHAP values 비교                                           ││
│  └─────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Data Platform IaC Structure                               │
│                                                                              │
│  terraform/                                                                  │
│  ├── environments/                                                           │
│  │   ├── dev/                                                               │
│  │   │   ├── main.tf                                                        │
│  │   │   ├── variables.tf                                                   │
│  │   │   └── terraform.tfvars                                               │
│  │   ├── staging/                                                           │
│  │   └── prod/                                                              │
│  │                                                                           │
│  ├── modules/                                                                │
│  │   ├── data-lake/                                                         │
│  │   │   ├── s3.tf           # S3 버킷, 라이프사이클                        │
│  │   │   ├── glue.tf         # Glue Catalog, Crawlers                       │
│  │   │   ├── lake-formation.tf                                              │
│  │   │   └── outputs.tf                                                     │
│  │   │                                                                       │
│  │   ├── streaming/                                                         │
│  │   │   ├── kinesis.tf      # Kinesis Streams, Firehose                   │
│  │   │   ├── msk.tf          # MSK Cluster                                  │
│  │   │   └── flink.tf        # Managed Flink                                │
│  │   │                                                                       │
│  │   ├── analytics/                                                         │
│  │   │   ├── redshift.tf     # Redshift Cluster/Serverless                 │
│  │   │   ├── athena.tf       # Athena Workgroups                           │
│  │   │   └── quicksight.tf   # QuickSight                                  │
│  │   │                                                                       │
│  │   └── governance/                                                        │
│  │       ├── iam.tf          # IAM Roles, Policies                         │
│  │       ├── kms.tf          # KMS Keys                                     │
│  │       └── monitoring.tf   # CloudWatch, Alarms                          │
│  │                                                                           │
│  └── shared/                                                                 │
│      ├── vpc.tf              # VPC, Subnets, Endpoints                      │
│      └── security-groups.tf                                                 │
└─────────────────────────────────────────────────────────────────────────────┘

// modules/data-lake/main.tf

# S3 Data Lake Buckets
resource "aws_s3_bucket" "data_lake" {
  for_each = toset(["raw", "curated", "analytics"])
  
  bucket = "${var.project_name}-${each.key}-${var.environment}"
  
  tags = merge(var.tags, {
    Layer = each.key
  })
}

resource "aws_s3_bucket_versioning" "data_lake" {
  for_each = aws_s3_bucket.data_lake
  bucket   = each.value.id
  
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "data_lake" {
  for_each = aws_s3_bucket.data_lake
  bucket   = each.value.id

  rule {
    id     = "transition-to-ia"
    status = "Enabled"

    transition {
      days          = 90
      storage_class = "STANDARD_IA"
    }

    transition {
      days          = 180
      storage_class = "GLACIER"
    }

    noncurrent_version_expiration {
      noncurrent_days = 30
    }
  }
}

# Glue Database
resource "aws_glue_catalog_database" "data_lake" {
  for_each = toset(["raw", "curated", "analytics"])
  
  name = "${var.project_name}_${each.key}_${var.environment}"
  
  create_table_default_permission {
    permissions = ["ALL"]
    
    principal {
      data_lake_principal_identifier = "IAM_ALLOWED_PRINCIPALS"
    }
  }
}

# Lake Formation Registration
resource "aws_lakeformation_resource" "data_lake" {
  for_each = aws_s3_bucket.data_lake
  
  arn = each.value.arn
}

# Glue Crawler
resource "aws_glue_crawler" "data_lake" {
  for_each = {
    raw      = { schedule = "cron(0 */6 * * ? *)", path = "raw/" }
    curated  = { schedule = "cron(0 */2 * * ? *)", path = "curated/" }
  }
  
  name          = "${var.project_name}-${each.key}-crawler"
  database_name = aws_glue_catalog_database.data_lake[each.key].name
  role          = var.glue_role_arn
  schedule      = each.value.schedule

  s3_target {
    path = "s3://${aws_s3_bucket.data_lake[each.key].id}/${each.value.path}"
  }

  schema_change_policy {
    delete_behavior = "LOG"
    update_behavior = "UPDATE_IN_DATABASE"
  }

  configuration = jsonencode({
    Version = 1.0
    Grouping = {
      TableGroupingPolicy = "CombineCompatibleSchemas"
    }
  })
}

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Data Pipeline Test Pyramid                                │
│                                                                              │
│                           ┌─────────────┐                                   │
│                           │   E2E Test  │  • 전체 파이프라인 검증           │
│                           │             │  • 프로덕션 유사 환경             │
│                           └─────────────┘  • 느림, 비용 높음                │
│                                                                              │
│                      ┌─────────────────────┐                                │
│                      │  Integration Test   │  • 서비스 간 연동 검증         │
│                      │                     │  • 실제 AWS 서비스 사용        │
│                      └─────────────────────┘  • 중간 속도/비용              │
│                                                                              │
│                 ┌───────────────────────────────┐                           │
│                 │      Data Quality Test        │  • 데이터 품질 규칙       │
│                 │                               │  • 스키마 검증            │
│                 └───────────────────────────────┘  • 빠름                   │
│                                                                              │
│            ┌─────────────────────────────────────────┐                      │
│            │           Unit Test                     │  • ETL 로직 검증     │
│            │                                         │  • 변환 함수 테스트  │
│            └─────────────────────────────────────────┘  • 가장 빠름         │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

// ETL 변환 함수

# etl/transformations.py
from pyspark.sql import DataFrame
from pyspark.sql.functions import col, when, lit, to_date

def clean_customer_data(df: DataFrame) -> DataFrame:
    """고객 데이터 정제"""
    return df \
        .filter(col("customer_id").isNotNull()) \
        .filter(col("email").rlike("^[a-zA-Z0-9+_.-]+@[a-zA-Z0-9.-]+$")) \
        .withColumn("status", 
            when(col("status").isNull(), lit("active"))
            .otherwise(col("status"))) \
        .dropDuplicates(["customer_id"])

def calculate_order_metrics(df: DataFrame) -> DataFrame:
    """주문 메트릭 계산"""
    return df \
        .withColumn("order_date", to_date(col("order_timestamp"))) \
        .withColumn("total_amount", col("quantity") * col("unit_price")) \
        .withColumn("discount_amount", 
            when(col("discount_rate").isNotNull(), 
                 col("total_amount") * col("discount_rate"))
            .otherwise(lit(0))) \
        .withColumn("final_amount", 
            col("total_amount") - col("discount_amount"))

// Great Expectations 설정

# expectations/orders_expectations.py
import great_expectations as gx

def create_orders_expectations(context):
    """주문 데이터 기대치 정의"""
    
    suite = context.add_expectation_suite("orders_suite")
    
    # 필수 컬럼 존재
    suite.add_expectation(
        gx.expectations.ExpectColumnToExist(column="order_id")
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnToExist(column="customer_id")
    )
    
    # Null 검사
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
    )
    
    # 고유성 검사
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
    )
    
    # 값 범위 검사
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="amount",
            min_value=0,
            max_value=1000000
        )
    )
    
    # 값 집합 검사
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="status",
            value_set=["pending", "completed", "cancelled", "refunded"]
        )
    )
    
    # 날짜 형식 검사
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToMatchStrftimeFormat(
            column="order_date",
            strftime_format="%Y-%m-%d"
        )
    )
    
    # 행 수 검사
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(
            min_value=1000,
            max_value=10000000
        )
    )
    
    return suite

# CI/CD에서 실행
def validate_data(df, context):
    """데이터 검증 실행"""
    results = context.run_validation_operator(
        "action_list_operator",
        assets_to_validate=[df],
        expectation_suite_name="orders_suite"
    )
    
    if not results["success"]:
        failed_expectations = [
            r for r in results["results"] 
            if not r["success"]
        ]
        raise ValueError(f"Data validation failed: {failed_expectations}")
    
    return results

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Data Pipeline CI/CD Flow                                  │
│                                                                              │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐   │
│  │  Source │───▶│  Build  │───▶│  Test   │───▶│ Staging │───▶│  Prod   │   │
│  │ (Git)   │    │(CodeBuild)   │(Quality)│    │ Deploy  │    │ Deploy  │   │
│  └─────────┘    └─────────┘    └─────────┘    └─────────┘    └─────────┘   │
│       │              │              │              │              │          │
│       ▼              ▼              ▼              ▼              ▼          │
│  • CodeCommit   • Unit Test    • Data Quality • Glue Job    • Manual       │
│  • GitHub       • Lint         • Schema       • Step Func   • Approval     │
│  • S3           • Package      • Integration  • Lambda      • Canary       │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

// Terraform CodePipeline 구성

resource "aws_codepipeline" "data_pipeline" {
  name     = "data-pipeline-cicd"
  role_arn = aws_iam_role.codepipeline.arn

  artifact_store {
    location = aws_s3_bucket.artifacts.bucket
    type     = "S3"
    
    encryption_key {
      id   = aws_kms_key.artifacts.arn
      type = "KMS"
    }
  }

  # Source Stage
  stage {
    name = "Source"
    
    action {
      name             = "Source"
      category         = "Source"
      owner            = "AWS"
      provider         = "CodeStarSourceConnection"
      version          = "1"
      output_artifacts = ["source_output"]
      
      configuration = {
        ConnectionArn    = aws_codestarconnections_connection.github.arn
        FullRepositoryId = "org/data-platform"
        BranchName       = "main"
      }
    }
  }

  # Build Stage
  stage {
    name = "Build"
    
    action {
      name             = "Build"
      category         = "Build"
      owner            = "AWS"
      provider         = "CodeBuild"
      input_artifacts  = ["source_output"]
      output_artifacts = ["build_output"]
      version          = "1"
      
      configuration = {
        ProjectName = aws_codebuild_project.build.name
      }
    }
  }

  # Test Stage
  stage {
    name = "Test"
    
    action {
      name            = "UnitTest"
      category        = "Test"
      owner           = "AWS"
      provider        = "CodeBuild"
      input_artifacts = ["build_output"]
      version         = "1"
      run_order       = 1
      
      configuration = {
        ProjectName = aws_codebuild_project.unit_test.name
      }
    }
    
    action {
      name            = "DataQualityTest"
      category        = "Test"
      owner           = "AWS"
      provider        = "CodeBuild"
      input_artifacts = ["build_output"]
      version         = "1"
      run_order       = 2
      
      configuration = {
        ProjectName = aws_codebuild_project.data_quality.name
      }
    }
  }

  # Staging Deploy
  stage {
    name = "Staging"
    
    action {
      name            = "DeployToStaging"
      category        = "Deploy"
      owner           = "AWS"
      provider        = "CloudFormation"
      input_artifacts = ["build_output"]
      version         = "1"
      
      configuration = {
        ActionMode    = "CREATE_UPDATE"
        StackName     = "data-pipeline-staging"
        TemplatePath  = "build_output::template.yaml"
        Capabilities  = "CAPABILITY_IAM,CAPABILITY_AUTO_EXPAND"
        RoleArn       = aws_iam_role.cloudformation.arn
        ParameterOverrides = jsonencode({
          Environment = "staging"
        })
      }
    }
  }

  # Production Deploy with Approval
  stage {
    name = "Production"
    
    action {
      name     = "ManualApproval"
      category = "Approval"
      owner    = "AWS"
      provider = "Manual"
      version  = "1"
      run_order = 1
      
      configuration = {
        NotificationArn = aws_sns_topic.approval.arn
        CustomData      = "Please review staging deployment before approving production"
      }
    }
    
    action {
      name            = "DeployToProduction"
      category        = "Deploy"
      owner           = "AWS"
      provider        = "CloudFormation"
      input_artifacts = ["build_output"]
      version         = "1"
      run_order       = 2
      
      configuration = {
        ActionMode    = "CREATE_UPDATE"
        StackName     = "data-pipeline-production"
        TemplatePath  = "build_output::template.yaml"
        Capabilities  = "CAPABILITY_IAM,CAPABILITY_AUTO_EXPAND"
        RoleArn       = aws_iam_role.cloudformation.arn
        ParameterOverrides = jsonencode({
          Environment = "production"
        })
      }
    }
  }
}

// buildspec.yml

version: 0.2

env:
  variables:
    PYTHON_VERSION: "3.9"
  secrets-manager:
    DB_PASSWORD: "prod/database:password"

phases:
  install:
    runtime-versions:
      python: 3.9
    commands:
      - pip install --upgrade pip
      - pip install -r requirements.txt
      - pip install pytest pytest-cov great_expectations

  pre_build:
    commands:
      - echo "Running linting..."
      - flake8 src/ --max-line-length=120
      - black --check src/
      - echo "Running security scan..."
      - bandit -r src/

  build:
    commands:
      - echo "Running unit tests..."
      - pytest tests/unit/ -v --cov=src --cov-report=xml
      - echo "Packaging Glue jobs..."
      - zip -r glue_jobs.zip src/glue/
      - aws s3 cp glue_jobs.zip s3://${ARTIFACT_BUCKET}/glue/${CODEBUILD_RESOLVED_SOURCE_VERSION}/

  post_build:
    commands:
      - echo "Generating CloudFormation template..."
      - python scripts/generate_cfn.py
      - echo "Build completed on $(date)"

reports:
  pytest_reports:
    files:
      - "coverage.xml"
    file-format: COBERTURAXML

artifacts:
  files:
    - template.yaml
    - parameters/*.json
  discard-paths: no

cache:
  paths:
    - '/root/.cache/pip/**/*'