Repository for dip

5.2 KiB

Raw Permalink Blame History

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를 참조하세요.

5.2 KiB Raw Permalink Blame History