service-catalog/manifests/helm/flink-cdc-job/0.1.0/CUSTOM-README.md

Flink CDC Job 배포

이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출하는 chart입니다.
Session Cluster를 재시작하지 않고 파이프라인만 독립적으로 배포/재배포할 수 있습니다.

사전 조건: flink-cdc-session chart로 Session Cluster가 RUNNING 상태여야 합니다.

1. 배포 방법

Release 이름이 Job, ConfigMap, ServiceAccount, Role 이름으로 사용됩니다.

helm upgrade --install cdc-job ./ \
  --namespace flink-test \
  -f values.yaml \
  -f custom-values.yaml

여러 파이프라인 동시 배포

# 사용자 테이블 파이프라인
helm install cdc-users ./ -n flink-test -f values.yaml -f values-users.yaml

# 주문 테이블 파이프라인
helm install cdc-orders ./ -n flink-test -f values.yaml -f values-orders.yaml

파이프라인 재제출

Kubernetes Job은 완료 후 재실행이 불가하므로 uninstall → install로 재제출합니다.

helm uninstall cdc-job -n flink-test
helm install cdc-job ./ -n flink-test -f values.yaml -f custom-values.yaml

2. custom-values.yaml 설명

1) ServiceAccount / RBAC

Name	설명	기본값
serviceAccount.create	ServiceAccount 생성 여부	true
serviceAccount.name	SA 이름. 비어있으면 Release.Name 사용	""
rbac.create	Role / RoleBinding 생성 여부	true

2) Session Cluster 연결

Name	설명	기본값
sessionCluster.restAddress	Session Cluster REST 서비스명	flink-cdc-session-session-rest
sessionCluster.restPort	REST 포트	8081

restAddress는 FlinkDeployment 이름에 -rest를 붙인 값입니다.
flink-cdc-session chart는 FlinkDeployment를 {release-name}-session으로 생성하므로
REST 서비스명은 {release-name}-session-rest가 됩니다.
예: release name flink-cdc-session → sessionCluster.restAddress: flink-cdc-session-session-rest

3) PostgreSQL 소스

Name	설명	기본값
postgres.hostname	PostgreSQL 호스트	postgres-postgresql.flink-test.svc.cluster.local
postgres.port	PostgreSQL 포트	5432
postgres.username	CDC 전용 사용자	flink_cdc
postgres.password	사용자 패스워드	password
postgres.slotName	Replication slot 이름	flink_cdc_slot
postgres.tables	CDC 대상 테이블 목록	testdb.public.orders,testdb.public.users

4) 파이프라인

Name	설명	기본값
pipeline.name	파이프라인 이름	cdc-pipeline
pipeline.parallelism	파이프라인 병렬도	2
pipeline.checkpointInterval	체크포인트 주기	60s

5) Iceberg 싱크

Name	설명	기본값
sink.catalog.uri	Lakekeeper REST URI	https://lakekeeper.example.org/catalog
sink.catalog.warehouse	카탈로그 warehouse 이름	minio
sink.catalog.s3Endpoint	Iceberg 데이터 저장소 엔드포인트	https://minio.example.org
sink.catalog.oauth2Uri	Keycloak OAuth2 토큰 URI	https://keycloak.example.org/…
sink.catalog.credential	client_id:client_secret	-
sink.catalog.scope	OAuth2 scope	lakekeeper

6) 라우팅

route:
  - sourceTable: public.orders      # PostgreSQL schema.table
    sinkTable: testdb_public.orders # Iceberg namespace.table
  - sourceTable: public.users
    sinkTable: testdb_public.users

3. values 파일 예시 (복수 파이프라인)

values-users.yaml (사용자/상품 테이블)

sessionCluster:
  restAddress: flink-cdc-session-session-rest

pipeline:
  name: cdc-users
  parallelism: 2
  checkpointInterval: 60s

postgres:
  hostname: postgres.example.svc.cluster.local
  port: 5432
  username: flink_cdc
  password: "password"
  slotName: flink_cdc_users_slot
  decodingPlugin: pgoutput
  tables: mydb.public.users,mydb.public.products

route:
  - sourceTable: public.users
    sinkTable: mydb_public.users
  - sourceTable: public.products
    sinkTable: mydb_public.products

values-orders.yaml (주문 테이블)

sessionCluster:
  restAddress: flink-cdc-session-session-rest

pipeline:
  name: cdc-orders
  parallelism: 4
  checkpointInterval: 60s

postgres:
  hostname: postgres.example.svc.cluster.local
  slotName: flink_cdc_orders_slot
  tables: mydb.public.orders,mydb.public.order_items

route:
  - sourceTable: public.orders
    sinkTable: mydb_public.orders
  - sourceTable: public.order_items
    sinkTable: mydb_public.order_items

4. 배포 후 확인

# Job 상태 확인
kubectl get job -n flink-test

# Job 로그 확인 (release name: cdc-job)
kubectl logs -n flink-test job/cdc-job

# Flink REST API로 Job 상태 확인
kubectl exec -n flink-test <jobmanager-pod> -- curl -s http://localhost:8081/jobs

5. 삭제

helm uninstall cdc-job -n flink-test

삭제 시 주의사항: cdc-job을 삭제하여도 Flink Session Cluster에 제출된 Job은 삭제되지 않습니다.

6. 참고

자세한 파라미터 설명 및 렌더링 확인 방법은 README.md를 참조하세요.