diff --git a/manifests/helm/flink-cdc-job/0.1.0/BUILD-README.md b/manifests/helm/flink-cdc-job/0.1.0/BUILD-README.md new file mode 100644 index 0000000..c04bd81 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/BUILD-README.md @@ -0,0 +1,103 @@ +# flink-cdc-job 버전 갱신 가이드 + +## 1. git 작업 환경 구성 + +- DIP 카탈로그 git 다운로드 +``` +$ git clone https://github.com/paasup/dip-catalog.git +``` + +- 작업 브랜치로 체크아웃 +``` +$ git checkout -b update-flink-cdc-job/0.1.0 +``` + +## 2. helm 차트 개발 및 업데이트 + +- 현재 flink-cdc-job 차트는 직접 작성하여 업데이트 중입니다. +- 기존 오픈소스 차트를 기반으로 하지 않고, 프로젝트 요구사항에 맞게 커스텀 개발하고 있습니다. + +### 2.1 차트 구조 +``` +flink-cdc-job/0.1.0/ +├── Chart.yaml # 차트 메타데이터 +├── values.yaml # 기본 설정값 +├── custom-values.yaml # 커스텀 설정값 +├── BUILD-README.md # 빌드 가이드 (본 파일) +├── CUSTOM-README.md # 배포 가이드 +├── README.md # 차트 상세 문서 +└── templates/ + ├── _helpers.tpl # 헬퍼 템플릿 + ├── configmap.yaml # 이 Job 전용 pipeline.yaml + └── job.yaml # Kubernetes Job (파이프라인 제출) +``` + +### 2.2 주요 기능 +- 기존 Flink Session Cluster에 CDC 파이프라인 Job을 독립적으로 제출 +- `flink-cdc-session` chart와 분리되어 클러스터 재시작 없이 파이프라인 배포/재배포 가능 +- 여러 Release 이름으로 복수의 파이프라인 동시 운용 +- Release 이름 기반 Job/ConfigMap 네이밍으로 이름 충돌 방지 + +## 3. 이미지 정보 + +- 기본 이미지: `wbsong111/flink-cdc-pipeline:3.5.0-flink1.20-r1` +- Flink CDC 3.5.0 + Flink 1.20 기반 커스텀 이미지 +- PostgreSQL CDC 커넥터 및 Iceberg 싱크 포함 + +## 4. git push 및 tag 추가 + +- 갱신작업 진행후 commit +```sh +$ git add . +$ git commit -m "update flink-cdc-job/0.1.0" +``` + +- main 브랜치에 체크아웃 후 merge +```sh +$ git checkout main +$ git merge update-flink-cdc-job/0.1.0 +``` + +- git에 push 후 작업 브랜치 삭제 +```sh +$ git push -u origin main +$ git branch -d update-flink-cdc-job/0.1.0 +``` + +- git tag 추가 후 push +```sh +$ git tag flink-cdc-job/0.1.0 +$ git push origin flink-cdc-job/0.1.0 +``` + +## 5. 차트 버전 정보 + +- flink-cdc-job/0.1.0 + - Apache Flink CDC 3.5.0 + Flink 1.20 기반 + - 이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출 + - PostgreSQL → Apache Iceberg REST 카탈로그 CDC 파이프라인 + - 여러 파이프라인을 독립 Release로 동시 운용 가능 + - 서비스 배포를 위하여 custom-values.yaml에 정의하였다. + - 차트의 빌드 방법과 배포 방법을 BUILD-README.md, CUSTOM-README.md 문서에 작성하였다. + +## 6. 개발 참고사항 + +### 6.1 flink-cdc-session과의 역할 분리 + +| 항목 | flink-cdc-session | flink-cdc-job | +|------|-------------------|---------------| +| FlinkDeployment (Session Cluster) | ✅ | ❌ | +| 파이프라인 ConfigMap | ✅ (클러스터용) | ✅ (Job 전용, 독립) | +| Job 제출 | 선택적 (`job.enabled`) | 항상 배포 | +| 배포 주기 | 클러스터 변경 시 | 파이프라인 변경 시 | + +### 6.2 주요 의존성 +- Flink Kubernetes Operator 설치 필요 +- `flink-cdc-session` chart로 Session Cluster가 실행 중이어야 함 +- Kubernetes 1.25+ 권장 +- Helm 3.x 필요 + +## 참고) 관련 문서 +- [Flink CDC 공식 문서](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/) +- [Flink Kubernetes Operator 문서](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/) +- [Apache Iceberg 문서](https://iceberg.apache.org/docs/latest/) diff --git a/manifests/helm/flink-cdc-job/0.1.0/CUSTOM-README.md b/manifests/helm/flink-cdc-job/0.1.0/CUSTOM-README.md new file mode 100644 index 0000000..9b80852 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/CUSTOM-README.md @@ -0,0 +1,155 @@ +# Flink CDC Job 배포 + +이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출하는 chart입니다. +Session Cluster를 재시작하지 않고 파이프라인만 독립적으로 배포/재배포할 수 있습니다. + +> **사전 조건**: `flink-cdc-session` chart로 Session Cluster가 RUNNING 상태여야 합니다. + +## 1. 배포 방법 + +Release 이름이 Job 및 ConfigMap 이름으로 사용됩니다. + +```sh +helm upgrade --install cdc-job ./ \ + --namespace flink-test \ + -f values.yaml \ + -f custom-values.yaml +``` + +### 여러 파이프라인 동시 배포 + +```sh +# 사용자 테이블 파이프라인 +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로 재제출합니다. + +```sh +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) Session Cluster 연결 + +| Name | 설명 | 기본값 | +|------|------|--------| +| `sessionCluster.restAddress` | Session Cluster REST 서비스명 | `flink-cdc-session-rest` | +| `sessionCluster.restPort` | REST 포트 | `8081` | + +> `restAddress`는 FlinkDeployment 이름에 `-rest`를 붙인 값입니다. +> 예: FlinkDeployment `flink-cdc-session` → 서비스명 `flink-cdc-session-rest` + +### 2) 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) 라우팅 + +```yaml +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 (사용자/상품 테이블) + +```yaml +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 (주문 테이블) + +```yaml +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. 배포 후 확인 + +```bash +# Job 상태 확인 +kubectl get job -n flink-test + +# Job 로그 확인 +kubectl logs -n flink-test job/cdc-job +``` + +## 5. 삭제 + +```bash +helm uninstall cdc-job -n flink-test +``` +> **삭제 시 주의사항** : `cdc-job`을 삭제하여도 flink session cluster에 제출된 job은 삭제되지 않습니다. + +## 6. 참고 + +자세한 파라미터 설명 및 렌더링 확인 방법은 [README.md](README.md)를 참조하세요. diff --git a/manifests/helm/flink-cdc-job/0.1.0/Chart.yaml b/manifests/helm/flink-cdc-job/0.1.0/Chart.yaml new file mode 100644 index 0000000..5560d5c --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/Chart.yaml @@ -0,0 +1,6 @@ +apiVersion: v2 +appVersion: 3.5.0-flink1.20 +description: Submits a Flink CDC pipeline job to an existing Flink Session Cluster +name: flink-cdc-job +type: application +version: 0.1.0 diff --git a/manifests/helm/flink-cdc-job/0.1.0/README.md b/manifests/helm/flink-cdc-job/0.1.0/README.md new file mode 100644 index 0000000..30c7690 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/README.md @@ -0,0 +1,214 @@ +# flink-cdc-job Helm Chart + +이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출하는 chart입니다. +`flink-cdc-session` chart와 분리되어 있으므로 Session Cluster를 재시작하지 않고 파이프라인만 독립적으로 배포/재배포할 수 있습니다. + +## flink-cdc-session과의 역할 분리 + +| 항목 | flink-cdc-session | flink-cdc-job | +|------|-------------------|---------------| +| FlinkDeployment (Session Cluster) | ✅ | ❌ | +| 파이프라인 ConfigMap | ✅ (클러스터용) | ✅ (Job 전용, 독립) | +| Job 제출 | 선택적 (`job.enabled`) | 항상 배포 | +| 배포 주기 | 클러스터 변경 시 | 파이프라인 변경 시 | + +## 사전 조건 + +- `flink-cdc-session` chart로 Session Cluster가 실행 중이어야 합니다. + +## 파일 구조 + +``` +flink-cdc-job/ +├── Chart.yaml +├── values.yaml +└── templates/ + ├── _helpers.tpl + ├── configmap.yaml # 이 Job 전용 pipeline.yaml (Release.Name 기반 네이밍) + └── job.yaml # Kubernetes Job +``` + +## 배포 + +### 기본 배포 + +```bash +helm install cdc-job ./flink/helm/flink-cdc-job -n flink-test +``` + +Release 이름이 Job 및 ConfigMap 이름으로 사용됩니다. +- Job: `cdc-job` +- ConfigMap: `cdc-job-config` + +### 여러 파이프라인 동시 배포 + +각 파이프라인을 독립 release로 설치합니다. + +```bash +# 사용자 테이블 파이프라인 +helm install cdc-users ./flink/helm/flink-cdc-job -n flink-test \ + -f values-users.yaml + +# 주문 테이블 파이프라인 +helm install cdc-orders ./flink/helm/flink-cdc-job -n flink-test \ + -f values-orders.yaml +``` + +### 파이프라인 재제출 + +Kubernetes Job은 완료 후 재실행이 불가하므로, 재제출 시 uninstall → install 합니다. + +```bash +helm uninstall cdc-job -n flink-test +helm install cdc-job ./flink/helm/flink-cdc-job -n flink-test +``` + +### 삭제 + +```bash +helm uninstall cdc-job -n flink-test +``` + +--- + +## values.yaml 파라미터 + +### 이미지 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `image.repository` | `wbsong111/flink-cdc-pipeline` | 컨테이너 이미지 레지스트리/이름 | +| `image.tag` | `3.5.0-flink1.20-r1` | 이미지 태그 | +| `image.pullPolicy` | `Always` | 이미지 풀 정책 | +| `serviceAccount` | `flink` | Kubernetes ServiceAccount 이름 | + +### Session Cluster 연결 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `sessionCluster.restAddress` | `flink-cdc-session-rest` | Session Cluster REST 서비스명 | +| `sessionCluster.restPort` | `8081` | REST 포트 | + +> `restAddress`는 FlinkDeployment 이름에 `-rest`를 붙인 값입니다. +> 예: FlinkDeployment `flink-cdc-session` → 서비스명 `flink-cdc-session-rest` + +### Flink CDC 글로벌 설정 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `flinkCdc.parallelism` | `1` | 전체 파이프라인 기본 병렬도 | +| `flinkCdc.schemaChangeBehavior` | `EVOLVE` | 스키마 변경 처리 방식 (`EVOLVE` / `IGNORE` / `EXCEPTION`) | + +### 파이프라인 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `pipeline.name` | `cdc-pipeline` | Flink Job 이름 (Flink UI 표시명) | +| `pipeline.parallelism` | `2` | 이 파이프라인의 병렬도 | +| `pipeline.checkpointInterval` | `60s` | 체크포인트 주기 | + +### PostgreSQL 소스 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `postgres.hostname` | `postgres-postgresql.flink-test.svc.cluster.local` | PostgreSQL 호스트 | +| `postgres.port` | `5432` | PostgreSQL 포트 | +| `postgres.username` | `flink_cdc` | CDC 전용 사용자 | +| `postgres.password` | `password` | 사용자 비밀번호 | +| `postgres.slotName` | `flink_cdc_slot` | Replication Slot 이름 (파이프라인마다 고유해야 함) | +| `postgres.decodingPlugin` | `pgoutput` | 디코딩 플러그인 (`pgoutput` / `decoderbufs`) | +| `postgres.tables` | `testdb.public.orders,testdb.public.users` | CDC 대상 테이블 목록 (`database.schema.table` 형식) | + +### Iceberg REST 카탈로그 싱크 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `sink.catalog.uri` | `https://lakekeeper.example.org/catalog` | Iceberg REST 카탈로그 URI | +| `sink.catalog.warehouse` | `minio` | 카탈로그 warehouse 이름 | +| `sink.catalog.s3Endpoint` | `https://minio.example.org` | S3 호환 스토리지 엔드포인트 | +| `sink.catalog.s3PathStyleAccess` | `"true"` | Path-style 접근 여부 | +| `sink.catalog.oauth2Uri` | `https://keycloak.example.org/...` | OAuth2 토큰 엔드포인트 | +| `sink.catalog.credential` | `lakekeeper-admin:...` | OAuth2 client_id:client_secret | +| `sink.catalog.scope` | `lakekeeper` | OAuth2 scope | + +### 라우팅 + +소스 테이블과 싱크 테이블 매핑입니다. + +```yaml +route: + - sourceTable: public.orders # PostgreSQL schema.table + sinkTable: testdb_public.orders # Iceberg namespace.table +``` + +--- + +## values 파일 예시 + +### values-users.yaml (사용자/상품 테이블) + +```yaml +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 (주문 테이블) + +```yaml +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 +``` + +--- + +## 렌더링 확인 + +```bash +helm template cdc-job . -n flink-test + +helm lint . + +# 여러 release 이름으로 ConfigMap/Job 이름이 분리되는지 확인 +helm template cdc-users . -n flink-test | grep "name:" +helm template cdc-orders . -n flink-test | grep "name:" +``` + +## 배포 후 확인 + +```bash +# Job 상태 확인 +kubectl get job -n flink-test + +# Job 로그 확인 +kubectl logs -n flink-test job/cdc-job +``` diff --git a/manifests/helm/flink-cdc-job/0.1.0/custom-values.yaml b/manifests/helm/flink-cdc-job/0.1.0/custom-values.yaml new file mode 100644 index 0000000..32cb6a2 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/custom-values.yaml @@ -0,0 +1,47 @@ +# custom-values.yaml +# 실제 배포에 사용하는 환경별 오버라이드 값 +# 사용법: helm install ./ -f values.yaml -f custom-values.yaml + +# 제출 대상 Session Cluster (flink-cdc-session chart의 FlinkDeployment 이름 + "-rest") +sessionCluster: + restAddress: flink-cdc-session-rest + restPort: 8081 + +# PostgreSQL 소스 연결 정보 +postgres: + hostname: postgres-postgresql.flink-test.svc.cluster.local + port: 5432 + username: flink_cdc + password: password + # 파이프라인마다 고유한 slot 이름 사용 + slotName: flink_cdc_slot + decodingPlugin: pgoutput + # CDC 대상 테이블 목록 (database.schema.table 형식, 쉼표 구분) + tables: testdb.public.orders,testdb.public.users + +# 파이프라인 설정 +pipeline: + name: cdc-pipeline + parallelism: 2 + checkpointInterval: 60s + +# Iceberg REST 카탈로그 싱크 +sink: + catalog: + uri: https://lakekeeper.example.org/catalog + warehouse: minio + s3Endpoint: https://minio.example.org + s3PathStyleAccess: "true" + oauth2Uri: https://keycloak.example.org/realms/paasup/protocol/openid-connect/token + # client_id:client_secret 형식 + credential: lakekeeper-admin:changeme + scope: lakekeeper + +# 소스→싱크 테이블 라우팅 +# sourceTable: PostgreSQL schema.table 형식 +# sinkTable: Iceberg namespace.table 형식 +route: + - sourceTable: public.orders + sinkTable: testdb_public.orders + - sourceTable: public.users + sinkTable: testdb_public.users diff --git a/manifests/helm/flink-cdc-job/0.1.0/templates/_helpers.tpl b/manifests/helm/flink-cdc-job/0.1.0/templates/_helpers.tpl new file mode 100644 index 0000000..5fc0033 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/templates/_helpers.tpl @@ -0,0 +1,26 @@ +{{/* +Chart label +*/}} +{{- define "flink-cdc-job.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Fully qualified name: Release.Name (chart name 중복 시 제거) +*/}} +{{- define "flink-cdc-job.fullname" -}} +{{- .Release.Name | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "flink-cdc-job.labels" -}} +helm.sh/chart: {{ include "flink-cdc-job.chart" . }} +app.kubernetes.io/name: {{ .Chart.Name }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} diff --git a/manifests/helm/flink-cdc-job/0.1.0/templates/configmap.yaml b/manifests/helm/flink-cdc-job/0.1.0/templates/configmap.yaml new file mode 100644 index 0000000..81d4bcb --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/templates/configmap.yaml @@ -0,0 +1,45 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "flink-cdc-job.fullname" . }}-config + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-job.labels" . | nindent 4 }} +data: + flink-cdc.yaml: | + parallelism: {{ .Values.flinkCdc.parallelism }} + schema.change.behavior: {{ .Values.flinkCdc.schemaChangeBehavior }} + + pipeline.yaml: | + source: + type: postgres + hostname: {{ .Values.postgres.hostname }} + port: {{ .Values.postgres.port }} + username: {{ .Values.postgres.username }} + password: {{ .Values.postgres.password }} + decoding.plugin.name: {{ .Values.postgres.decodingPlugin }} + slot.name: {{ .Values.postgres.slotName }} + tables: {{ .Values.postgres.tables }} + + pipeline: + name: {{ .Values.pipeline.name }} + parallelism: {{ .Values.pipeline.parallelism }} + execution.checkpointing.interval: {{ .Values.pipeline.checkpointInterval }} + + sink: + type: iceberg + catalog.properties.type: rest + catalog.properties.uri: {{ .Values.sink.catalog.uri | quote }} + catalog.properties.warehouse: {{ .Values.sink.catalog.warehouse }} + catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO + catalog.properties.s3.endpoint: {{ .Values.sink.catalog.s3Endpoint | quote }} + catalog.properties.s3.path-style-access: {{ .Values.sink.catalog.s3PathStyleAccess | quote }} + catalog.properties.oauth2-server-uri: {{ .Values.sink.catalog.oauth2Uri | quote }} + catalog.properties.credential: {{ .Values.sink.catalog.credential }} + catalog.properties.scope: {{ .Values.sink.catalog.scope }} + + route: + {{- range .Values.route }} + - source-table: {{ .sourceTable }} + sink-table: {{ .sinkTable }} + {{- end }} diff --git a/manifests/helm/flink-cdc-job/0.1.0/templates/job.yaml b/manifests/helm/flink-cdc-job/0.1.0/templates/job.yaml new file mode 100644 index 0000000..1bfb0be --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/templates/job.yaml @@ -0,0 +1,33 @@ +apiVersion: batch/v1 +kind: Job +metadata: + name: {{ include "flink-cdc-job.fullname" . }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-job.labels" . | nindent 4 }} +spec: + backoffLimit: 0 + template: + spec: + serviceAccountName: {{ .Values.serviceAccount }} + containers: + - name: submit + image: {{ .Values.image.repository }}:{{ .Values.image.tag }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + command: ["/bin/sh", "-c"] + args: + - | + cd /opt/flink/flink-cdc-3.5.0 + ./bin/flink-cdc.sh \ + ./conf/pipeline.yaml \ + -Dexecution.checkpointing.interval={{ .Values.pipeline.checkpointInterval }} \ + -Drest.address={{ .Values.sessionCluster.restAddress }} \ + -Drest.port={{ .Values.sessionCluster.restPort }} + volumeMounts: + - name: pipeline-config + mountPath: /opt/flink/flink-cdc-3.5.0/conf + volumes: + - name: pipeline-config + configMap: + name: {{ include "flink-cdc-job.fullname" . }}-config + restartPolicy: Never diff --git a/manifests/helm/flink-cdc-job/0.1.0/values.yaml b/manifests/helm/flink-cdc-job/0.1.0/values.yaml new file mode 100644 index 0000000..89ceab5 --- /dev/null +++ b/manifests/helm/flink-cdc-job/0.1.0/values.yaml @@ -0,0 +1,53 @@ +image: + repository: wbsong111/flink-cdc-pipeline + tag: 3.5.0-flink1.20-r1 + pullPolicy: Always + +serviceAccount: flink + +# 제출 대상 Session Cluster REST 엔드포인트 +# flink-cdc-session chart가 생성한 서비스명: {FlinkDeployment.metadata.name}-rest +sessionCluster: + restAddress: flink-cdc-session-rest + restPort: 8081 + +# Flink CDC 글로벌 설정 (flink-cdc.yaml) +flinkCdc: + parallelism: 1 + schemaChangeBehavior: EVOLVE + +# PostgreSQL 소스 +postgres: + hostname: postgres-postgresql.flink-test.svc.cluster.local + port: 5432 + username: flink_cdc + password: password + slotName: flink_cdc_slot + decodingPlugin: pgoutput + tables: testdb.public.orders,testdb.public.users + +# 파이프라인 설정 +pipeline: + name: cdc-pipeline + parallelism: 2 + checkpointInterval: 60s + +# Iceberg REST 카탈로그 싱크 +sink: + catalog: + uri: https://lakekeeper.example.org/catalog + warehouse: minio + s3Endpoint: https://minio.example.org + s3PathStyleAccess: "true" + oauth2Uri: https://keycloak.example.org/realms/paasup/protocol/openid-connect/token + credential: lakekeeper-admin:wMyE9WNn50DH8yycLXwLTmMmu66JU7GV + scope: lakekeeper + +# 소스→싱크 테이블 라우팅 +# source-table: PostgreSQL CDC emit 형식 (schema.table) +# sink-table: Iceberg namespace.table 형식 +route: + - sourceTable: public.orders + sinkTable: testdb_public.orders + - sourceTable: public.users + sinkTable: testdb_public.users