[DE 7편] Airflow와 AWS S3 연동 기초: Terraform 인프라 프로비저닝 & LocalFilesystemToS3Operator 실습

1. 왜 클라우드 데이터 레이크의 중심으로 Amazon S3를 사용하는가?

온프레미스 파일 서버나 단일 데이터베이스(RDBMS)에 방대한 원천 데이터를 계속 적재하면 디스크 용량 한계와 I/O 병목이 발생합니다.
Amazon S3 (Simple Storage Service)는 다음과 같은 이유로 현대 데이터 엔지니어링의 핵심 데이터 레이크(Data Lake) 저장소로 사용됩니다:

• 99.999999999% (11 9s)의 압도적인 데이터 내구성
• 무제한 객체 스토리지 용량 및 자동 확장성
• 강력한 빅데이터 생태계 연계 (AWS Athena, Glue, EMR/Spark, Redshift, Snowflake 등)
• 저렴한 비용과 유연한 수명주기(Lifecycle) 티어링 정책

이번 8편에서는 Terraform(IaC)을 통해 보안 모범 사례가 적용된 S3 버킷을 프로비저닝하고, Airflow에서 LocalFilesystemToS3Operator와 S3Hook을 사용하여 데이터를 안전하게 업로드하고 검증하는 파이프라인을 구축합니다.

---

2. Terraform(IaC)으로 안전한 S3 버킷 프로비저닝 (infra/s3.tf)

실무에서는 AWS 콘솔에서 수동으로 버킷을 생성하지 않고, 재현 가능성과 보안 표준을 준수하기 위해 Terraform을 사용합니다.

소스 코드 위치: infra/s3.tf

# S3 버킷 전역 고유 이름 정의
locals {
  # S3 버킷 네이밍 규칙: 3~63자, 소문자, 숫자, 하이픈(-)만 허용
  # 글로벌 고유성을 확보하기 위해 AWS Account ID를 접미사로 결합
  airflow_bucket_name = "${var.project_name}-s3-bk-${data.aws_caller_identity.current.account_id}"
}

# 1. S3 버킷 기본 리소스 생성
resource "aws_s3_bucket" "airflow_data" {
  bucket        = local.airflow_bucket_name
  # false: 버킷 내부에 파일(Object)이 남아있으면 terraform destroy 방지 (데이터 유실 방지)
  # true : 개발/테스트 환경에서 버킷 내부 파일까지 일괄 삭제 허용
  force_destroy = var.s3_force_destroy

  tags = merge(
    local.common_tags,
    {
      Name = local.airflow_bucket_name
    }
  )
}

# 2. 객체 소유권 강제 (Object Ownership - ACL 비활성화)
resource "aws_s3_bucket_ownership_controls" "airflow_data" {
  bucket = aws_s3_bucket.airflow_data.id

  rule {
    # BucketOwnerEnforced: 레거시 ACL을 비활성화하고 버킷 소유자가 모든 객체를 소유 (권장 보안 표준)
    object_ownership = "BucketOwnerEnforced"
  }
}

# 3. 퍼블릭 액세스 차단 (Public Access Block)
# 데이터 레이크는 외부 인터넷 노출을 원천 차단하고 private으로 관리해야 합니다.
resource "aws_s3_bucket_public_access_block" "airflow_data" {
  bucket = local.airflow_bucket_name

  block_public_acls       = true
  ignore_public_acls      = true
  block_public_policy     = true
  restrict_public_buckets = true
}

# 4. 서버 측 기본 암호화 설정 (SSE-S3 AES-256)
resource "aws_s3_bucket_server_side_encryption_configuration" "airflow_data" {
  bucket = local.airflow_bucket_name

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "AES256"
    }
  }
}

🔒 엔터프라이즈 S3 보안 3대 원칙:

1. Public Access Block: 실수로 데이터가 외부에 유출되는 보안 사고를 방지합니다.
2. BucketOwnerEnforced: 버킷 소유자가 모든 객체 권한을 독점하여 IAM Policy로만 제어합니다.
3. Server-Side Encryption (SSE-S3): 저장되는 모든 파일(At-Rest)을 자동 암호화합니다.

---

3. Airflow AWS 연동 환경 준비

1) Amazon Provider 패키지 설치

Airflow에서 S3 관련 Operator와 Hook을 사용하려면 apache-airflow-providers-amazon 패키지가 필요합니다.

pip install apache-airflow-providers-amazon

Docker Compose 환경에서는 _PIP_ADDITIONAL_REQUIREMENTS에 apache-airflow-providers-amazon을 추가하거나 커스텀 Dockerfile에 포함시킵니다.

2) Airflow Connection (aws_default) 등록

Airflow Web UI 메인 메뉴 ➔ Admin ➔ Connections에서 새 연결을 생성합니다:

설정 항목	값	설명
Connection Id	aws_default	DAG 코드에서 참조할 연결 식별자
Connection Type	Amazon Web Services	AWS Provider 타입
AWS Access Key ID	AKIA...	IAM 사용자 액세스 키
AWS Secret Access Key	wJalrX...	IAM 사용자 시크릿 키
Extra	{"region_name": "ap-northeast-2"}	기본 리전 (서울 리전)

---

4. Airflow S3 업로드 및 검증 DAG 구현 (dags/08_aws_s3_basic.py)

소스 코드 위치: dags/08_aws_s3_basic.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.transfers.local_to_s3 import LocalFilesystemToS3Operator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from datetime import datetime, timedelta
import logging
import pendulum

# 1. 환경 설정
KST = pendulum.timezone("Asia/Seoul")
BUCKET_NAME = "de-ai-25-infra-s3-bk-827913617635"  # 생성한 S3 버킷명
UPLOAD_FILE_NAME = "costs.csv"
LOCAL_PATH = f"/opt/airflow/dags/data/{UPLOAD_FILE_NAME}"

# Task 2: S3Hook을 통한 실제 업로드 여부 검증
def _check_s3(**kwargs):
    # 1. S3Hook 인스턴스 생성 (aws_default 커넥션 활용)
    hook = S3Hook(aws_conn_id="aws_default")

    # 2. 버킷 내의 모든 Object Key 조회
    keys = hook.list_keys(bucket_name=BUCKET_NAME)

    if not keys:
        raise ValueError(f"버킷 [{BUCKET_NAME}] 내에 파일이 존재하지 않습니다. 업로드 실패!")

    # 3. 특정 파일이 정상 업로드되었는지 확인
    if UPLOAD_FILE_NAME in keys:
        logging.info(f"✅ S3 버킷 내 [{UPLOAD_FILE_NAME}] 파일 업로드 검증 완료!")
    else:
        raise FileNotFoundError(f"버킷 내에 [{UPLOAD_FILE_NAME}] 키를 찾을 수 없습니다.")

# 2. DAG 정의
with DAG(
    dag_id="08_aws_s3_basic",
    description="로컬 데이터를 AWS S3 버킷으로 단순 업로드 및 검증하는 기본 DAG",
    default_args={
        "owner": "aic-de1-admin",
        "retries": 1,
        "retry_delay": timedelta(minutes=1),
    },
    schedule_interval="@daily",
    start_date=pendulum.datetime(2026, 6, 29, tz=KST),
    catchup=False,
    tags=["aws", "s3", "storage"]
) as dag:

    # 1. LocalFilesystemToS3Operator: 로컬 파일을 S3 객체로 전송
    task_upload_to_s3 = LocalFilesystemToS3Operator(
        task_id="upload_to_s3",
        filename=LOCAL_PATH,         # 컨테이너 로컬 상의 원본 파일 절대 경로
        dest_key=UPLOAD_FILE_NAME,   # S3 버킷 내 저장될 객체 Key (파일명)
        dest_bucket=BUCKET_NAME,      # 대상 S3 버킷 이름
        aws_conn_id="aws_default",    # 사용할 AWS 커넥션 ID
        replace=True                 # 동일한 키가 이미 존재할 경우 덮어쓰기 허용
    )

    # 2. PythonOperator: S3Hook을 활용한 객체 존재 검증
    task_check_s3 = PythonOperator(
        task_id="check_s3",
        python_callable=_check_s3
    )

    # 의존성 연결
    task_upload_to_s3 >> task_check_s3

---

5.  데이터 엔지니어 실무 현업 꿀팁 (Tip)

1. S3 Key 파티셔닝 네이밍 전략 (Hive Partitioning)

S3에 파일을 단순히 루트에 쌓아두면 파일 수가 수만 개를 넘어갈 때 리스팅 성능이 급격히 저하되고 Athena 쿼리 비용이 폭증합니다.

• 권장 구조: s3://bucket/service_name/year=YYYY/month=MM/day=DD/filename.csv
• 이렇게 key=value 형태의 Hive 파티셔닝을 적용하면, 후속 데이터 분석 도구(Athena, Spark, Presto)가 Partition Pruning을 통해 해당 날짜 폴더만 스캔하므로 쿼리 속도는 수십 배 빨라지고 비용은 90% 이상 절감됩니다.

2. IAM 최소 권한 원칙 (Principle of Least Privilege)

Airflow 서비스용 IAM 사용자에게 AdministratorAccess나 AmazonS3FullAccess를 부여하는 것은 위험합니다.

• Airflow 전용 IAM Policy를 생성하여 대상 버킷에 한해 s3:PutObject, s3:GetObject, s3:ListBucket, s3:DeleteObject 권한만 최소 부여하세요.
• AWS EKS / EC2 환경이라면 하드코딩된 Access Key 대신 IAM Roles for Service Accounts (IRSA) 또는 EC2 Instance Profile을 사용하여 키 유출 위험을 원천 차단하세요.

3. 대용량 파일 전송 시 Multipart Upload 최적화

기가바이트(GB) 이상의 대용량 로그나 덤프 파일을 업로드할 때는 단일 스트림 전송 시 네트워크 끊김으로 전체 재전송이 발생할 수 있습니다.

• S3Hook.load_file() 내부적으로 boto3의 TransferConfig가 동작하여 Multipart Upload(멀티파트 병렬 전송)가 지원되지만, 파일 크기가 매우 큰 경우 multipart_threshold 및 max_concurrency 값을 튜닝하여 네트워크 대역폭을 극대화할 수 있습니다.

[!TIP]

4. S3 Lifecycle Rule(수명 주기 규칙)을 통한 비용 절감

데이터 레이크에 저장된 원천 데이터(Bronze Layer)는 시간이 지나면 조회 빈도가 급격히 감소합니다.

• 30일 경과: Standard ➔ S3 Standard-IA (Infrequent Access)로 전환
• 90일 경과: S3 Glacier Flexible Retrieval / Deep Archive로 자동 아카이빙
• 1년 경과: 규정상 보관 의무가 없는 임시 파일 자동 삭제 (Expiration)

Terraform aws_s3_bucket_lifecycle_configuration 리소스로 위 규칙을 코드화해두면 클라우드 비용을 획기적으로 절약할 수 있습니다.

---

6. 7편 요약 및 다음 편 예고

이번 7편에서는 Terraform을 통해 엔터프라이즈 보안 표준이 적용된 AWS S3 버킷을 프로비저닝하고, Airflow의 LocalFilesystemToS3Operator와 S3Hook을 통해 데이터를 클라우드 데이터 레이크에 안전하게 적재 및 검증하는 방법을 다루었습니다.

8편에서는 실무에서 널리 쓰이는 이벤트 기반(Event-Driven) 데이터 파이프라인을 완성합니다. 센서 데이터를 주기적으로 쏘아 올리는 Producer DAG와, S3KeySensor(mode="reschedule")를 통해 파일 도착을 실시간 감지하여 자동 파싱 및 정리하는 Consumer DAG를 구축해보겠습니다.

 

 

오늘의 한 마디

매번 하는 말이고 이미 작성했을 수도 있지만 결국 확률이 있는 것을 하는 것이 맞다고 생각한다. 자기가 하고 싶은게 있으면 그것을 할 수 있게 되는 확률이 있는 행동을 해야한다.

연애를 하고 싶으면 연애를 할 수 있는 환경에 뛰어들어야한다. 만약 당신이 10억을 받을 확률이 0% 이고 1억을 받을 수 있는 확률이 1%라면 어디에 베팅을 하겠습니까? 당연히 1%라도 걸고 그쪽으로 가야합니다.

그러니 제 말은 조금이라도 확률이 있는 곳에 가서 뭐라도 해야한다는 말입니다. 그리고 점점 그 확률을 높이는 것을 목표로 해야합니다.