You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
175 lines
5.2 KiB
175 lines
5.2 KiB
# Flink CDC Job 배포
|
|
|
|
이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출하는 chart입니다.
|
|
Session Cluster를 재시작하지 않고 파이프라인만 독립적으로 배포/재배포할 수 있습니다.
|
|
|
|
> **사전 조건**: `flink-cdc-session` chart로 Session Cluster가 RUNNING 상태여야 합니다.
|
|
|
|
## 1. 배포 방법
|
|
|
|
Release 이름이 Job, ConfigMap, ServiceAccount, Role 이름으로 사용됩니다.
|
|
|
|
```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) 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) 라우팅
|
|
|
|
```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
|
|
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 (주문 테이블)
|
|
|
|
```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. 배포 후 확인
|
|
|
|
```bash
|
|
# 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. 삭제
|
|
|
|
```bash
|
|
helm uninstall cdc-job -n flink-test
|
|
```
|
|
|
|
> **삭제 시 주의사항**: `cdc-job`을 삭제하여도 Flink Session Cluster에 제출된 Job은 삭제되지 않습니다.
|
|
|
|
## 6. 참고
|
|
|
|
자세한 파라미터 설명 및 렌더링 확인 방법은 [README.md](README.md)를 참조하세요.
|
|
|