MongoDB to S3 Parquet ETL (Application Repository)

이 저장소는 MongoDB 데이터를 추출해 Parquet 파일로 변환한 뒤 Amazon S3에 업로드하는 Python 기반 ETL 애플리케이션을 관리합니다. Kubernetes 및 Argo CD 매니페스트는 별도 GitOps 저장소(alcha-gitops)에서 관리합니다.

프로젝트 구조

• app/
  • config.py – 환경 변수로부터 설정을 읽고 검증합니다.
  • etl.py – MongoDB → Parquet → S3 업로드 파이프라인의 엔트리 포인트입니다.
• requirements.txt – 런타임 의존성 목록입니다.
• Dockerfile – 배포용 컨테이너 이미지를 빌드하는 정의입니다.

로컬 개발

1. 가상환경 생성 및 의존성 설치

   python -m venv .venv
   source .venv/bin/activate
   pip install -r requirements.txt

2. 환경 변수 설정 후 실행

   export MONGO_URI="mongodb://user:pass@host:27017/?authSource=admin"
   export MONGO_DATABASE="vehicle_telemetry"
   export MONGO_COLLECTION="realtime_data"
   export S3_BUCKET="team-alcha-etl-stage"
   export S3_PREFIX="curated/realtime"
   export AWS_ACCESS_KEY_ID="your-access-key"
   export AWS_SECRET_ACCESS_KEY="your-secret-key"
   export AWS_REGION="ap-northeast-2"
   export LOOKBACK_SECONDS="600"
   export OUTPUT_FILENAME_TEMPLATE="{sequence}.snappy.parquet"
   export S3_KEY_TEMPLATE="{prefix}year={timestamp_obj:%Y}/month={timestamp_obj:%m}/day={timestamp_obj:%d}/hour={timestamp_obj:%H}/{filename}"
   export STATE_S3_KEY="state/realtime-watermark.json"
   export FILENAME_SEQUENCE_WIDTH="5"
   export OUTPUT_TIMEZONE="UTC"
   
   python -m app.etl

   필요 시 MONGO_QUERY, MONGO_PROJECTION, S3_PREFIX, OUTPUT_FILENAME_TEMPLATE 등을 추가로 지정할 수 있습니다.

3. 코드 품질 점검(선택)

   pip install black isort mypy pylint
   black app
   isort app
   pylint app
   mypy app

컨테이너 이미지 빌드 및 푸시

GitLab CI/CD 파이프라인 또는 로컬에서 다음 명령을 사용해 이미지를 빌드할 수 있습니다.

docker build -t <registry>/mongo-s3-parquet-etl:<tag> .
docker push <registry>/mongo-s3-parquet-etl:<tag>

권장되는 CI 단계는 다음과 같습니다.

1. 의존성 설치 및 단위 테스트 실행
2. Lint/타입체크
3. docker build 후 docker push
4. GitOps 저장소(alcha-gitops)의 CronJob 매니페스트 이미지 태그 업데이트

환경 변수 요약

이름	설명	필수	기본값
MONGO_URI	MongoDB 접속 URI	✔	-
MONGO_DATABASE	대상 데이터베이스	✔	-
MONGO_COLLECTION	대상 컬렉션	✔	-
MONGO_QUERY	추출 조건(JSON 문자열)		{}
MONGO_PROJECTION	필드 프로젝션(JSON 문자열)		None
MONGO_BATCH_SIZE	커서 배치 크기		1000
LOOKBACK_SECONDS	최근 N초 범위 제한		None
S3_BUCKET	업로드 대상 버킷	✔	-
S3_PREFIX	업로드 경로 prefix		``
S3_KEY_TEMPLATE	파티셔닝/파일명 템플릿 (prefix, collection, timestamp, timestamp_obj, sequence, partition_key, filename 사용 가능)		None
STATE_S3_BUCKET	워터마크 상태 버킷		None (=S3_BUCKET)
STATE_S3_KEY	워터마크 상태 키		None
OUTPUT_FILENAME_TEMPLATE	파일명 템플릿 (sequence 등 context 사용 가능)		{collection}-{timestamp}.parquet
OUTPUT_TIMESTAMP_FORMAT	타임스탬프 포맷		%Y%m%dT%H%M%SZ
OUTPUT_TIMEZONE	파티션 생성에 사용할 타임존		UTC
PARQUET_COMPRESSION	Parquet 압축 방식		snappy
AWS_REGION	AWS 리전		환경변수 또는 IAM 설정
FILENAME_SEQUENCE_WIDTH	파일명 sequence 자리수		5

MongoDB 문서의 timestamp 필드는 UTC 기준으로 해석되며, 시간 단위(floor('H'))로 잘라 S3 경로 템플릿의 timestamp/timestamp_obj 값으로 전달됩니다. 한 시간 구간(UTC)에 속한 문서는 하나의 Parquet 파일로 합쳐지며, S3_KEY_TEMPLATE에서 날짜·시간 파티션을 구성하면 Redshift가 이벤트 발생 시점 기준으로 데이터를 조회할 수 있습니다.

GitOps 저장소와의 연계

• Kubernetes 관련 매니페스트는 alcha-gitops 저장소의 connector/mongo-s3-parquet-etl/ 디렉터리에서 관리합니다.
• 새 이미지를 빌드하면 GitOps 저장소에서 CronJob 이미지 태그를 갱신하고 Argo CD 동기화를 수행해 배포합니다.
• Secrets Manager, IRSA, Cron 스케줄 등 인프라 관련 변경도 GitOps 저장소에서 관리하세요.