From bb69f903f40866694ad5477ea83f1108d7f2e00a Mon Sep 17 00:00:00 2001 From: wbsong111 Date: Fri, 23 Jan 2026 08:17:33 +0900 Subject: [PATCH] Add chart flink-sql-gateway --- manifests/helm/flink-sql-gateway/Chart.yaml | 22 + manifests/helm/flink-sql-gateway/README.md | 411 ++++++++++++++++++ .../helm/flink-sql-gateway/custom-values.yaml | 53 +++ .../flink-sql-gateway/templates/_helpers.tpl | 107 +++++ .../templates/configmaps.yaml | 148 +++++++ .../templates/flink-session-cluster.yaml | 67 +++ .../templates/flink-sql-gateway.yaml | 159 +++++++ .../flink-sql-gateway/templates/rbac.yaml | 49 +++ .../flink-sql-gateway/templates/services.yaml | 71 +++ .../templates/sql-client.yaml | 83 ++++ manifests/helm/flink-sql-gateway/values.yaml | 212 +++++++++ 11 files changed, 1382 insertions(+) create mode 100644 manifests/helm/flink-sql-gateway/Chart.yaml create mode 100644 manifests/helm/flink-sql-gateway/README.md create mode 100644 manifests/helm/flink-sql-gateway/custom-values.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/_helpers.tpl create mode 100644 manifests/helm/flink-sql-gateway/templates/configmaps.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/flink-session-cluster.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/flink-sql-gateway.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/rbac.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/services.yaml create mode 100644 manifests/helm/flink-sql-gateway/templates/sql-client.yaml create mode 100644 manifests/helm/flink-sql-gateway/values.yaml diff --git a/manifests/helm/flink-sql-gateway/Chart.yaml b/manifests/helm/flink-sql-gateway/Chart.yaml new file mode 100644 index 0000000..0d3ebf4 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/Chart.yaml @@ -0,0 +1,22 @@ +apiVersion: v2 +name: flink-sql-gateway +description: A Helm chart for Flink SQL Gateway with Kafka integration +type: application +version: 0.1.0 +appVersion: "2.0.1" +keywords: + - flink + - sql + - gateway + - kafka + - streaming +home: https://github.com/your-org/flink-sql-gateway-helm +sources: + - https://github.com/your-org/flink-sql-gateway-helm +maintainers: + - name: Flink SQL Gateway Team + email: team@example.com +dependencies: [] +annotations: + category: Analytics + licenses: Apache-2.0 diff --git a/manifests/helm/flink-sql-gateway/README.md b/manifests/helm/flink-sql-gateway/README.md new file mode 100644 index 0000000..8852c18 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/README.md @@ -0,0 +1,411 @@ +# Flink SQL Gateway Helm Chart + +Apache Flink SQL Gateway를 쉽게 배포하기 위한 Helm Chart입니다. 외부 Python 클라이언트 접근에 최적화되어 있습니다. + +## 📋 개요 + +이 Helm Chart는 다음 구성 요소를 배포합니다: + +### 필수 컴포넌트 (항상 배포됨) +- **Flink Session Cluster**: SQL Gateway가 연결할 Flink 클러스터 +- **Flink SQL Gateway**: SQL 쿼리를 실행할 수 있는 REST API 서버 +- **Service**: SQL Gateway 접근을 위한 Kubernetes 서비스 +- **RBAC**: Flink 리소스 관리를 위한 권한 설정 +- **ConfigMap**: SQL Gateway 설정 + +### 선택적 컴포넌트 (설정으로 활성화) +- **SQL Client Pod**: 대화형 SQL 클라이언트 (`sqlClient.enabled: true`) +- **Kafka Integration**: Kafka 연동 설정 (`kafka.enabled: true`) +- **SQL Scripts ConfigMap**: SQL 스크립트 모음 (SQL Client 활성화 시) + +## 🏗️ 아키텍처 + +### 외부 Python 클라이언트 접근 (권장) +``` +┌──────────────────┐ +│ Python Client │ +│ (External) │ +└──────────────────┘ + │ + ▼ HTTP REST API +┌──────────────────┐ +│ SQL Gateway │ +│ (REST API) │ +│ Port: 8083 │ +└──────────────────┘ + │ + ▼ SQL 실행 요청 +┌─────────────────────┐ +│ Flink Session │ +│ Cluster │ +│ (Data Processing) │ +└─────────────────────┘ + │ + ▼ 데이터 처리 (선택적) +┌─────────────────┐ +│ Kafka Cluster │ +│ (Streaming) │ +└─────────────────┘ +``` + +### 내부 SQL Client 사용 (선택적) +``` +┌──────────────────┐ +│ SQL Client │ +│ (Pod) │ +└──────────────────┘ + │ + ▼ SQL 쿼리 전송 +┌──────────────────┐ +│ SQL Gateway │ +│ (REST API) │ +└──────────────────┘ + │ + ▼ SQL 실행 요청 +┌─────────────────────┐ +│ Flink Session │ +│ Cluster │ +└─────────────────────┘ +``` + +**주요 특징:** +- **외부 Python 클라이언트 최적화**: Port Forward를 통한 직접 REST API 접근 +- **유연한 구성**: 필요한 컴포넌트만 선택적 배포 +- **Session Mode**: 대화형 SQL 개발 환경 제공 +- **검증된 안정성**: YAML 파싱 오류 해결 및 템플릿 검증 완료 + +## 🚀 빠른 시작 + +### 1. 사전 요구사항 + +- Kubernetes 클러스터 +- Helm 3.x +- Flink Kubernetes Operator 설치됨 +- Strimzi Kafka Operator 설치됨 (Kafka 사용 시) + +### 2. 설치 + +#### 기본 설치 (외부 Python 클라이언트용 - 권장) +```bash +# 최소 리소스로 기본 설치 +helm install flink-sql-gateway ./ \ + --namespace flink-sql-gateway \ + --create-namespace + +# 외부 Python 클라이언트용 최적화 설치 (권장) +helm install flink-sql-gateway ./ \ + --namespace flink-sql-gateway \ + --create-namespace \ + --values ./custom-values.yaml +``` + +#### 고급 설치 옵션 +```bash +# SQL Client 포함 설치 (대화형 SQL 사용) +helm install flink-sql-gateway ./ \ + --namespace flink-sql-gateway \ + --create-namespace \ + --set sqlClient.enabled=true + +# Kafka 연동 포함 설치 +helm install flink-sql-gateway ./ \ + --namespace flink-sql-gateway \ + --create-namespace \ + --set kafka.enabled=true \ + --set kafka.user.password="your-kafka-password" + +# 완전한 환경 설치 (모든 컴포넌트) +helm install flink-sql-gateway ./ \ + --namespace flink-sql-gateway \ + --create-namespace \ + --values ./custom-values.yaml \ + --set sqlClient.enabled=true \ + --set kafka.enabled=true \ + --set kafka.user.password="your-kafka-password" +``` + +### 3. 외부 Python 클라이언트 접근 (권장) + +#### Port Forward 설정 +```bash +# SQL Gateway API 접근 +kubectl port-forward -n flink-sql-gateway svc/flink-sql-gateway 8083:8083 + +# Flink Web UI 접근 (선택적) +kubectl port-forward -n flink-sql-gateway svc/flink-session-cluster-rest 8081:8081 +``` + +#### Python 클라이언트 예제 +```python +import requests +import json + +# SQL Gateway 연결 +gateway_url = "http://localhost:8083" + +# 1. 세션 생성 +response = requests.post(f"{gateway_url}/v1/sessions") +session_handle = response.json()["sessionHandle"] +print(f"Session created: {session_handle}") + +# 2. SQL 실행 +sql_request = { + "statement": "SHOW TABLES" +} +response = requests.post( + f"{gateway_url}/v1/sessions/{session_handle}/statements", + json=sql_request +) +operation_handle = response.json()["operationHandle"] + +# 3. 결과 조회 +result_response = requests.get( + f"{gateway_url}/v1/sessions/{session_handle}/operations/{operation_handle}/result/0" +) +print("Query result:", result_response.json()) +``` + +### 4. 내부 SQL Client 접근 (선택적) + +```bash +# SQL Client에 연결 (sqlClient.enabled=true인 경우) +kubectl exec -it flink-sql-client -n flink-sql-gateway -- \ + /opt/flink/bin/sql-client.sh gateway \ + --endpoint http://flink-sql-gateway:8083 +``` + +## ⚙️ 설정 + +### 주요 설정 항목 + +| 파라미터 | 설명 | 기본값 | custom-values.yaml | +|----------|------|--------|-------------------| +| `global.namespace` | 배포할 네임스페이스 | `flink-sql-gateway` | `flink-sql-gateway` | +| `global.image.repository` | Flink 이미지 저장소 | `paasup/flink-sql` | `paasup/flink-sql` | +| `global.image.tag` | Flink 이미지 태그 | `1.20-kafka` | `1.20-kafka` | +| `sessionCluster.enabled` | Session Cluster 활성화 | `true` | `true` | +| `sqlGateway.enabled` | SQL Gateway 활성화 | `true` | `true` | +| `sqlClient.enabled` | SQL Client 활성화 | `false` | `false` | +| `kafka.enabled` | Kafka 연동 활성화 | `false` | `false` | + +### 리소스 설정 비교 + +#### 기본 설정 (values.yaml) +```yaml +# 최소 리소스 - 테스트용 +sessionCluster: + jobManager: + resources: + memory: 1024m + cpu: 0.5 + taskManager: + resources: + memory: 2048m + cpu: 1 + flinkConfiguration: + taskmanager.numberOfTaskSlots: "2" + +sqlGateway: + resources: + requests: + memory: 512Mi + cpu: 0.25 + limits: + memory: 1Gi + cpu: 0.5 +``` + +#### 최적화 설정 (custom-values.yaml) +```yaml +# 외부 Python 클라이언트용 최적화 - 프로덕션 권장 +sessionCluster: + jobManager: + resources: + memory: 2048m # 2배 증가 + cpu: 1 # 2배 증가 + taskManager: + resources: + memory: 4096m # 2배 증가 + cpu: 2 # 2배 증가 + flinkConfiguration: + taskmanager.numberOfTaskSlots: "4" # 2배 증가 + +sqlGateway: + resources: + requests: + memory: 1Gi # 2배 증가 + cpu: 0.5 # 2배 증가 + limits: + memory: 2Gi # 2배 증가 + cpu: 1 # 2배 증가 + +# SQL Client 비활성화 (외부 Python 클라이언트 사용) +sqlClient: + enabled: false +``` + +### Kafka 설정 (선택적) + +```yaml +kafka: + enabled: true + bootstrapServers: "kafka-cluster-kafka-external-bootstrap.kafka.svc.cluster.local:9094" + topics: + input: + name: test.input + output: + name: test.output + user: + name: flink-sql-gateway + password: "your-kafka-password" + # 또는 External Secret 사용 + externalSecret: + enabled: true + secretName: "kafka-user-credentials" + usernameKey: "username" + passwordKey: "password" +``` + +### External Secret 설정 + +```yaml +kafka: + user: + externalSecret: + enabled: true + secretName: "flink-sql-gateway" # Kafka namespace의 기존 secret + usernameKey: "username" + passwordKey: "password" +``` + +## 📝 사용 예시 + +### 1. 테이블 생성 + +```sql +-- 소스 테이블 생성 +CREATE TABLE test_input ( + id STRING, + message STRING, + timestamp_field TIMESTAMP(3), + WATERMARK FOR timestamp_field AS timestamp_field - INTERVAL '5' SECOND +) WITH ( + 'connector' = 'kafka', + 'topic' = 'test.input', + 'properties.bootstrap.servers' = 'kafka-cluster-kafka-external-bootstrap.kafka.svc.cluster.local:9094', + 'properties.group.id' = 'flink-sql-gateway', + 'properties.security.protocol' = 'SASL_SSL', + 'properties.sasl.mechanism' = 'SCRAM-SHA-512', + 'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username="flink-sql-gateway" password="YOUR_PASSWORD";', + 'properties.ssl.truststore.location' = '/opt/flink/certs/truststore.jks', + 'properties.ssl.truststore.password' = 'changeit', + 'properties.ssl.truststore.type' = 'JKS', + 'format' = 'json', + 'json.timestamp-format.standard' = 'ISO-8601' +); +``` + +### 2. SQL 쿼리 실행 + +```sql +-- 간단한 필터링 쿼리 +INSERT INTO test_output +SELECT + id, + CONCAT('Processed: ', message) as processed_message, + timestamp_field as original_timestamp, + CURRENT_TIMESTAMP as processing_time +FROM test_input +WHERE LENGTH(message) > 5; +``` + +## 🔧 관리 + +### 업그레이드 + +```bash +helm upgrade flink-sql-gateway . \ + --namespace flink-sql-gateway \ + --values values.yaml +``` + +### 제거 + +```bash +helm uninstall flink-sql-gateway --namespace flink-sql-gateway +``` + +### 상태 확인 + +```bash +# Helm 릴리스 상태 +helm status flink-sql-gateway -n flink-sql-gateway + +# Pod 상태 +kubectl get pods -n flink-sql-gateway + +# FlinkDeployment 상태 +kubectl get flinkdeployment -n flink-sql-gateway + +# 로그 확인 +kubectl logs -n flink-sql-gateway -l app=flink-sql-gateway +``` + +## 🐛 문제 해결 + +### 일반적인 문제 + +1. **SQL Gateway 연결 실패** + ```bash + # Gateway 상태 확인 + kubectl get deployment flink-sql-gateway -n flink-sql-gateway + kubectl logs -n flink-sql-gateway -l app=flink-sql-gateway + + # Port Forward 확인 + kubectl port-forward -n flink-sql-gateway svc/flink-sql-gateway 8083:8083 + curl http://localhost:8083/v1/info + ``` + +2. **Python 클라이언트 연결 실패** + ```python + # 연결 테스트 + import requests + try: + response = requests.get("http://localhost:8083/v1/info", timeout=5) + print("SQL Gateway 연결 성공:", response.json()) + except requests.exceptions.ConnectionError: + print("Port Forward가 실행 중인지 확인하세요") + except requests.exceptions.Timeout: + print("SQL Gateway가 시작되지 않았을 수 있습니다") + ``` + +3. **Flink Session Cluster 시작 실패** + ```bash + # FlinkDeployment 상태 확인 + kubectl describe flinkdeployment flink-session-cluster -n flink-sql-gateway + ``` + +4. **Kafka 연결 오류** + - Kafka 사용자 비밀번호 확인 + - 인증서 설정 확인 + - 네트워크 연결 확인 + +### 디버깅 명령어 + +```bash +# 전체 환경 상태 확인 +kubectl get all -n flink-sql-gateway + +# 이벤트 확인 +kubectl get events -n flink-sql-gateway --sort-by='.lastTimestamp' + +# Helm 템플릿 확인 +helm template flink-sql-gateway . --values values.yaml +``` + +## 📚 참고 자료 + +- [Apache Flink Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.20/) +- [Flink SQL Gateway Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql-gateway/) +- [Flink Kubernetes Operator](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-1.13/) +- [Helm Documentation](https://helm.sh/docs/) diff --git a/manifests/helm/flink-sql-gateway/custom-values.yaml b/manifests/helm/flink-sql-gateway/custom-values.yaml new file mode 100644 index 0000000..56672df --- /dev/null +++ b/manifests/helm/flink-sql-gateway/custom-values.yaml @@ -0,0 +1,53 @@ +# Custom Values for Flink SQL Gateway Helm Chart +# Configuration for external Python client access with resource settings + +global: + namespace: flink-sql-test + image: + repository: wbsong111/flink-sql + tag: 2.0.1 + pullPolicy: IfNotPresent + + +# Flink Session Cluster resource configuration +sessionCluster: + flinkVersion: v2_0 + jobManager: + resources: + memory: 2048m # Increased from default 1024m + cpu: 1 # Increased from default 0.5 + taskManager: + resources: + memory: 4096m # Increased from default 2048m + cpu: 2 # Increased from default 1 + flinkConfiguration: + taskmanager.numberOfTaskSlots: "4" # Increased from default "2" + volumeMounts: + - name: kafka-certs + mountPath: /opt/flink/certs + readOnly: true + volumes: + - name: kafka-certs + secret: + secretName: kafka-ca-cert + +# SQL Gateway resource configuration +sqlGateway: + resources: + requests: + memory: 1Gi # Increased from default 512Mi + cpu: 0.5 # Increased from default 0.25 + limits: + memory: 2Gi # Increased from default 1Gi + cpu: 1 # Increased from default 0.5 + +sqlClient: + enabled: true +# Optional: Enable Ingress for external access +# ingress: +# enabled: true +# hosts: +# - host: flink-sql-gateway.example.com +# paths: +# - path: / +# pathType: Prefix diff --git a/manifests/helm/flink-sql-gateway/templates/_helpers.tpl b/manifests/helm/flink-sql-gateway/templates/_helpers.tpl new file mode 100644 index 0000000..5c3d129 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/_helpers.tpl @@ -0,0 +1,107 @@ +{{/* +Expand the name of the chart. +*/}} +{{- define "flink-sql-gateway.name" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Create a default fully qualified app name. +We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec). +If release name contains chart name it will be used as a full name. +*/}} +{{- define "flink-sql-gateway.fullname" -}} +{{- if .Values.fullnameOverride }} +{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- $name := default .Chart.Name .Values.nameOverride }} +{{- if contains $name .Release.Name }} +{{- .Release.Name | trunc 63 | trimSuffix "-" }} +{{- else }} +{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} +{{- end }} +{{- end }} +{{- end }} + +{{/* +Create chart name and version as used by the chart label. +*/}} +{{- define "flink-sql-gateway.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "flink-sql-gateway.labels" -}} +helm.sh/chart: {{ include "flink-sql-gateway.chart" . }} +{{ include "flink-sql-gateway.selectorLabels" . }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- with .Values.global.labels }} +{{ toYaml . }} +{{- end }} +{{- end }} + +{{/* +Selector labels +*/}} +{{- define "flink-sql-gateway.selectorLabels" -}} +app.kubernetes.io/name: {{ include "flink-sql-gateway.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end }} + +{{/* +Create the name of the service account to use +*/}} +{{- define "flink-sql-gateway.serviceAccountName" -}} +{{- if .Values.rbac.create }} +{{- default (include "flink-sql-gateway.fullname" .) .Values.rbac.serviceAccountName }} +{{- else }} +{{- default "default" .Values.rbac.serviceAccountName }} +{{- end }} +{{- end }} + +{{/* +Create namespace name +*/}} +{{- define "flink-sql-gateway.namespace" -}} +{{- default .Values.global.namespace .Release.Namespace }} +{{- end }} + +{{/* +Create image name +*/}} +{{- define "flink-sql-gateway.image" -}} +{{- printf "%s:%s" .Values.global.image.repository .Values.global.image.tag }} +{{- end }} + + +{{/* +Session Cluster labels +*/}} +{{- define "flink-sql-gateway.sessionCluster.labels" -}} +{{ include "flink-sql-gateway.labels" . }} +app: {{ .Values.sessionCluster.name }} +component: flink-session-cluster +{{- end }} + +{{/* +SQL Gateway labels +*/}} +{{- define "flink-sql-gateway.sqlGateway.labels" -}} +{{ include "flink-sql-gateway.labels" . }} +app: {{ .Values.sqlGateway.name }} +component: flink-sql-gateway +{{- end }} + +{{/* +SQL Client labels +*/}} +{{- define "flink-sql-gateway.sqlClient.labels" -}} +{{ include "flink-sql-gateway.labels" . }} +app: {{ .Values.sqlClient.name }} +component: flink-sql-client +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/configmaps.yaml b/manifests/helm/flink-sql-gateway/templates/configmaps.yaml new file mode 100644 index 0000000..5b6dca7 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/configmaps.yaml @@ -0,0 +1,148 @@ +--- +# ConfigMap for SQL Gateway Configuration +apiVersion: v1 +kind: ConfigMap +metadata: + name: sql-gateway-config + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +data: + sql-gateway-defaults.yaml: | + # SQL Gateway Configuration + sql-gateway: + endpoint: + rest: + address: {{ .Values.sqlGateway.config.endpoint.rest.address }} + port: {{ .Values.sqlGateway.config.endpoint.rest.port }} + bind-address: {{ .Values.sqlGateway.config.endpoint.rest.bindAddress }} + session: + max-num: {{ .Values.sqlGateway.config.session.maxNum }} + idle-timeout: {{ .Values.sqlGateway.config.session.idleTimeout }} + check-interval: {{ .Values.sqlGateway.config.session.checkInterval }} + plan-cache: + enabled: {{ .Values.sqlGateway.config.session.planCache.enabled }} + max-size: {{ .Values.sqlGateway.config.session.planCache.maxSize }} + ttl: {{ .Values.sqlGateway.config.session.planCache.ttl }} +{{- if .Values.sqlClient.enabled }} +--- +# ConfigMap with sample SQL scripts +apiVersion: v1 +kind: ConfigMap +metadata: + name: sql-test-scripts + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +data: + 01-create-tables.sql: | + -- Example SQL script for creating tables + -- This is a template - modify according to your data sources + + -- Example: Create source table (replace with your connector) + CREATE TABLE source_table ( + id STRING, + message STRING, + timestamp_field TIMESTAMP(3), + WATERMARK FOR timestamp_field AS timestamp_field - INTERVAL '5' SECOND + ) WITH ( + 'connector' = 'your-connector', -- e.g., 'kafka', 'filesystem', etc. + -- Add your connector-specific properties here + 'format' = 'json' + ); + + -- Example: Create sink table (replace with your connector) + CREATE TABLE sink_table ( + id STRING, + processed_message STRING, + original_timestamp TIMESTAMP(3), + processing_time TIMESTAMP(3) + ) WITH ( + 'connector' = 'your-connector', -- e.g., 'kafka', 'filesystem', etc. + -- Add your connector-specific properties here + 'format' = 'json' + ); + + 02-simple-query.sql: | + -- Simple filtering and transformation query + INSERT INTO sink_table + SELECT + id, + CONCAT('Processed: ', message) as processed_message, + timestamp_field as original_timestamp, + CURRENT_TIMESTAMP as processing_time + FROM source_table + WHERE LENGTH(message) > 5; + + 03-windowed-aggregation.sql: | + -- Windowed aggregation example + SELECT + TUMBLE_START(timestamp_field, INTERVAL '1' MINUTE) as window_start, + TUMBLE_END(timestamp_field, INTERVAL '1' MINUTE) as window_end, + COUNT(*) as message_count, + COUNT(DISTINCT id) as unique_ids + FROM source_table + GROUP BY TUMBLE(timestamp_field, INTERVAL '1' MINUTE); + + test-data-generator.sql: | + -- Generate test data (for manual testing) + -- This would typically be done by external producer + -- Example JSON format for input data: + -- {"id": "msg001", "message": "Hello World", "timestamp_field": "2024-12-10T10:00:00.000Z"} + -- {"id": "msg002", "message": "Test Message", "timestamp_field": "2024-12-10T10:01:00.000Z"} + + README.md: | + # Flink SQL Client Usage + + ## Connect to SQL Gateway with Remote Cluster + ```bash + # Connect to SQL Gateway and specify remote cluster + kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- /opt/flink/bin/sql-client.sh gateway \ + --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }} \ + -Dexecution.target=remote \ + -Drest.address={{ .Values.sessionCluster.name }}-rest \ + -Drest.port=8081 + ``` + + ## Alternative: Set execution config in SQL + ```bash + # Connect to SQL Gateway + kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- /opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }} + + # Then in SQL Client, set execution config: + SET 'execution.target' = 'remote'; + SET 'rest.address' = '{{ .Values.sessionCluster.name }}-rest'; + SET 'rest.port' = '8081'; + ``` + + ## Run SQL Scripts + ```bash + # Inside the SQL Client container + kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- bash + + # View available scripts + ls /opt/sql-scripts/ + + # Execute SQL file with remote cluster config + /opt/flink/bin/sql-client.sh gateway \ + --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }} \ + -Dexecution.target=remote \ + -Drest.address={{ .Values.sessionCluster.name }}-rest \ + -Drest.port=8081 \ + -f /opt/sql-scripts/01-create-tables.sql + ``` + + ## Test Data + Send JSON messages to your input source: + ```json + {"id": "msg001", "message": "Hello World", "timestamp_field": "2024-12-10T10:00:00.000Z"} + ``` +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/flink-session-cluster.yaml b/manifests/helm/flink-sql-gateway/templates/flink-session-cluster.yaml new file mode 100644 index 0000000..ef8e673 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/flink-session-cluster.yaml @@ -0,0 +1,67 @@ +{{- if .Values.sessionCluster.enabled }} +--- +# Flink Session Cluster (for SQL Gateway to connect to) +apiVersion: flink.apache.org/v1beta1 +kind: FlinkDeployment +metadata: + name: {{ .Values.sessionCluster.name }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.sessionCluster.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + image: {{ include "flink-sql-gateway.image" . }} + flinkVersion: {{ .Values.sessionCluster.flinkVersion }} + flinkConfiguration: + {{- toYaml .Values.sessionCluster.flinkConfiguration | nindent 4 }} + serviceAccount: {{ include "flink-sql-gateway.serviceAccountName" . }} + jobManager: + resource: + memory: {{ .Values.sessionCluster.jobManager.resources.memory }} + cpu: {{ .Values.sessionCluster.jobManager.resources.cpu }} + replicas: {{ .Values.sessionCluster.jobManager.replicas }} + taskManager: + resource: + memory: {{ .Values.sessionCluster.taskManager.resources.memory }} + cpu: {{ .Values.sessionCluster.taskManager.resources.cpu }} + replicas: {{ .Values.sessionCluster.taskManager.replicas }} + podTemplate: + spec: + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: flink-main-container + {{- with .Values.securityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.sessionCluster.env }} + env: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.sessionCluster.volumeMounts }} + volumeMounts: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.sessionCluster.volumes }} + volumes: + {{- toYaml . | nindent 8 }} + {{- end }} +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/flink-sql-gateway.yaml b/manifests/helm/flink-sql-gateway/templates/flink-sql-gateway.yaml new file mode 100644 index 0000000..f32035b --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/flink-sql-gateway.yaml @@ -0,0 +1,159 @@ +{{- if .Values.sqlGateway.enabled }} +--- +# SQL Gateway Deployment (separate from Flink cluster) +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ .Values.sqlGateway.name }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.sqlGateway.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + replicas: {{ .Values.sqlGateway.replicas }} + selector: + matchLabels: + app: {{ .Values.sqlGateway.name }} + template: + metadata: + labels: + app: {{ .Values.sqlGateway.name }} + {{- include "flink-sql-gateway.selectorLabels" . | nindent 8 }} + spec: + serviceAccountName: {{ include "flink-sql-gateway.serviceAccountName" . }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: sql-gateway + image: {{ include "flink-sql-gateway.image" . }} + imagePullPolicy: {{ .Values.global.image.pullPolicy }} + {{- with .Values.securityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + command: ["/bin/bash"] + args: + - -c + - | + # Wait for Flink session cluster to be ready with timeout + echo "Waiting for Flink session cluster..." + TIMEOUT=300 # 5 minutes + ELAPSED=0 + INTERVAL=5 + + while [ $ELAPSED -lt $TIMEOUT ]; do + if curl -f --connect-timeout 5 --max-time 10 http://{{ .Values.sessionCluster.name }}-rest:8081/v1/overview >/dev/null 2>&1; then + echo "Flink session cluster is ready!" + break + fi + echo "Waiting for Flink cluster... (${ELAPSED}s/${TIMEOUT}s)" + sleep $INTERVAL + ELAPSED=$((ELAPSED + INTERVAL)) + done + + if [ $ELAPSED -ge $TIMEOUT ]; then + echo "ERROR: Timeout waiting for Flink session cluster" + echo "Checking cluster status..." + kubectl get pods -n {{ include "flink-sql-gateway.namespace" . }} -l app={{ .Values.sessionCluster.name }} || true + exit 1 + fi + + echo "Flink cluster is ready. Starting SQL Gateway..." + + # Create flink-conf.yaml with remote cluster configuration + cat > /opt/flink/conf/config.yaml << EOF + # Remote Flink Cluster Configuration + rest.address: {{ .Values.sessionCluster.name }}-rest + rest.port: 8081 + execution.target: remote + + # SQL Gateway Configuration + sql-gateway.endpoint.rest.address: 0.0.0.0 + sql-gateway.endpoint.rest.port: {{ .Values.sqlGateway.port }} + + # Table/SQL Configuration + table.exec.source.idle-timeout: 30s + table.exec.resource.default-parallelism: 1 + EOF + + # Start SQL Gateway + /opt/flink/bin/sql-gateway.sh start + + echo "SQL Gateway started successfully" + + # Wait for log file to be created and then tail it + LOG_FILE="" + for i in {1..30}; do + LOG_FILE=$(find /opt/flink/log -name "flink--sql-gateway-*.log" 2>/dev/null | head -1) + if [ -n "$LOG_FILE" ]; then + echo "Found log file: $LOG_FILE" + break + fi + echo "Waiting for log file to be created... ($i/30)" + sleep 2 + done + + if [ -n "$LOG_FILE" ]; then + echo "Tailing SQL Gateway log file..." + tail -f "$LOG_FILE" + else + echo "Log file not found, keeping container alive..." + # Keep container running without tailing logs + while true; do + echo "SQL Gateway is running on port {{ .Values.sqlGateway.port }}" + sleep 60 + done + fi + env: + - name: FLINK_CONF_DIR + value: "/opt/flink/conf" + {{- with .Values.sqlGateway.env }} + {{- toYaml . | nindent 12 }} + {{- end }} + ports: + - name: rest + containerPort: {{ .Values.sqlGateway.port }} + protocol: TCP + volumeMounts: + {{- with .Values.sqlGateway.volumeMounts }} + {{- toYaml . | nindent 12 }} + {{- end }} + - name: sql-gateway-config + mountPath: /opt/flink/conf/sql-gateway-defaults.yaml + subPath: sql-gateway-defaults.yaml + readOnly: true + resources: + {{- toYaml .Values.sqlGateway.resources | nindent 12 }} + {{- with .Values.sqlGateway.livenessProbe }} + livenessProbe: + {{- toYaml . | nindent 12 }} + {{- end }} + {{- with .Values.sqlGateway.readinessProbe }} + readinessProbe: + {{- toYaml . | nindent 12 }} + {{- end }} + volumes: + {{- with .Values.sqlGateway.volumes }} + {{- toYaml . | nindent 8 }} + {{- end }} + - name: sql-gateway-config + configMap: + name: sql-gateway-config +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/rbac.yaml b/manifests/helm/flink-sql-gateway/templates/rbac.yaml new file mode 100644 index 0000000..ce9ab51 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/rbac.yaml @@ -0,0 +1,49 @@ +{{- if .Values.rbac.create }} +--- +# ServiceAccount for Flink SQL Gateway +apiVersion: v1 +kind: ServiceAccount +metadata: + name: {{ include "flink-sql-gateway.serviceAccountName" . }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +--- +# ClusterRole for Flink SQL Gateway +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: {{ include "flink-sql-gateway.fullname" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} +rules: + - apiGroups: [""] + resources: ["pods", "services", "endpoints", "persistentvolumeclaims", "events", "configmaps", "secrets"] + verbs: ["*"] + - apiGroups: ["apps"] + resources: ["deployments", "replicasets"] + verbs: ["*"] + - apiGroups: ["flink.apache.org"] + resources: ["flinkdeployments", "flinksessionjobs"] + verbs: ["*"] +--- +# ClusterRoleBinding for Flink SQL Gateway +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: {{ include "flink-sql-gateway.fullname" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: {{ include "flink-sql-gateway.fullname" . }} +subjects: + - kind: ServiceAccount + name: {{ include "flink-sql-gateway.serviceAccountName" . }} + namespace: {{ include "flink-sql-gateway.namespace" . }} +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/services.yaml b/manifests/helm/flink-sql-gateway/templates/services.yaml new file mode 100644 index 0000000..466c10b --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/services.yaml @@ -0,0 +1,71 @@ +{{- if .Values.sqlGateway.enabled }} +--- +# Service for SQL Gateway +apiVersion: v1 +kind: Service +metadata: + name: {{ .Values.sqlGateway.name }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.sqlGateway.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + type: {{ .Values.services.sqlGateway.type }} + ports: + - name: rest + port: {{ .Values.services.sqlGateway.port }} + targetPort: {{ .Values.services.sqlGateway.targetPort }} + protocol: TCP + selector: + app: {{ .Values.sqlGateway.name }} +{{- end }} + +{{- if .Values.ingress.enabled }} +--- +# Ingress for SQL Gateway +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: {{ include "flink-sql-gateway.fullname" . }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.labels" . | nindent 4 }} + {{- with .Values.ingress.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + {{- if .Values.ingress.className }} + ingressClassName: {{ .Values.ingress.className }} + {{- end }} + {{- if .Values.ingress.tls }} + tls: + {{- range .Values.ingress.tls }} + - hosts: + {{- range .hosts }} + - {{ . | quote }} + {{- end }} + secretName: {{ .secretName }} + {{- end }} + {{- end }} + rules: + {{- range .Values.ingress.hosts }} + - host: {{ .host | quote }} + http: + paths: + {{- range .paths }} + - path: {{ .path }} + {{- if .pathType }} + pathType: {{ .pathType }} + {{- end }} + backend: + service: + name: {{ $.Values.sqlGateway.name }} + port: + number: {{ $.Values.services.sqlGateway.port }} + {{- end }} + {{- end }} +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/templates/sql-client.yaml b/manifests/helm/flink-sql-gateway/templates/sql-client.yaml new file mode 100644 index 0000000..20f30a8 --- /dev/null +++ b/manifests/helm/flink-sql-gateway/templates/sql-client.yaml @@ -0,0 +1,83 @@ +{{- if .Values.sqlClient.enabled }} +--- +# SQL Client Pod for interactive SQL queries +apiVersion: v1 +kind: Pod +metadata: + name: {{ .Values.sqlClient.name }} + namespace: {{ include "flink-sql-gateway.namespace" . }} + labels: + {{- include "flink-sql-gateway.sqlClient.labels" . | nindent 4 }} + {{- with .Values.global.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + serviceAccountName: {{ include "flink-sql-gateway.serviceAccountName" . }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 4 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 4 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 4 }} + {{- end }} + {{- with .Values.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 4 }} + {{- end }} + containers: + - name: sql-client + image: {{ include "flink-sql-gateway.image" . }} + imagePullPolicy: {{ .Values.global.image.pullPolicy }} + {{- with .Values.securityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} + command: ["/bin/bash"] + args: + - -c + - | + echo "Flink SQL Client Ready" + echo "Connect to SQL Gateway: {{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}" + echo "Usage: /opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}" + + # Wait for SQL Gateway to be ready + echo "Waiting for SQL Gateway..." + until curl -f http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}/v1/info; do + echo "Waiting for SQL Gateway..." + sleep 10 + done + + echo "SQL Gateway is ready!" + echo "You can now connect using:" + echo "/opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}" + + # Keep container running + tail -f /dev/null + {{- with .Values.sqlClient.env }} + env: + {{- toYaml . | nindent 8 }} + {{- end }} + volumeMounts: + {{- with .Values.sqlClient.volumeMounts }} + {{- toYaml . | nindent 8 }} + {{- end }} + - name: sql-scripts + mountPath: /opt/sql-scripts + readOnly: true + resources: + {{- toYaml .Values.sqlClient.resources | nindent 8 }} + volumes: + {{- with .Values.sqlClient.volumes }} + {{- toYaml . | nindent 4 }} + {{- end }} + - name: sql-scripts + configMap: + name: sql-test-scripts + restartPolicy: Always +{{- end }} \ No newline at end of file diff --git a/manifests/helm/flink-sql-gateway/values.yaml b/manifests/helm/flink-sql-gateway/values.yaml new file mode 100644 index 0000000..059fb5b --- /dev/null +++ b/manifests/helm/flink-sql-gateway/values.yaml @@ -0,0 +1,212 @@ +# Default values for flink-sql-gateway. +# This is a YAML-formatted file. +# Declare variables to be passed into your templates. + +# Global settings +global: + namespace: flink-sql-test + image: + repository: paasup/flink-sql-gateway + tag: 1.20-kafka + pullPolicy: IfNotPresent + labels: + app: flink-sql-gateway + version: v1.20.0 + annotations: {} + +# Flink Session Cluster configuration +sessionCluster: + enabled: true + name: flink-session-cluster + flinkVersion: v1_20 + jobManager: + replicas: 1 + resources: + memory: 1024m + cpu: 0.5 + taskManager: + replicas: 1 + resources: + memory: 2048m + cpu: 1 + flinkConfiguration: + taskmanager.numberOfTaskSlots: "2" + parallelism.default: "1" + execution.checkpointing.storage.fs.path: file:///tmp/flink-checkpoints + execution.savepoint.path: file:///tmp/flink-savepoints + execution.checkpointing.interval: 60s + table.exec.source.idle-timeout: 30s + table.exec.resource.default-parallelism: "1" + # Optional environment variables for the session cluster + env: [] + # Example: + # env: + # - name: KAFKA_BOOTSTRAP_SERVERS + # value: "kafka-cluster:9092" + + # Optional volume mounts for the session cluster + volumeMounts: [] + # Example: + # volumeMounts: + # - name: kafka-certs + # mountPath: /opt/flink/certs + # readOnly: true + + # Optional volumes for the session cluster + volumes: [] + # Example: + # volumes: + # - name: kafka-certs + # secret: + # secretName: kafka-cluster-ca-cert + +# SQL Gateway configuration +sqlGateway: + enabled: true + name: flink-sql-gateway + replicas: 1 + port: 8083 + resources: + requests: + memory: 512Mi + cpu: 0.25 + limits: + memory: 1Gi + cpu: 0.5 + livenessProbe: + httpGet: + path: /v1/info + port: 8083 + initialDelaySeconds: 60 + periodSeconds: 30 + readinessProbe: + httpGet: + path: /v1/info + port: 8083 + initialDelaySeconds: 30 + periodSeconds: 10 + config: + endpoint: + rest: + address: 0.0.0.0 + port: 8083 + bindAddress: 0.0.0.0 + session: + maxNum: 1000 + idleTimeout: 600000 # 10 minutes + checkInterval: 60000 # 1 minute + planCache: + enabled: true + maxSize: 100 + ttl: 3600000 # 1 hour + # Optional environment variables for the SQL Gateway + env: [] + # Example: + # env: + # - name: KAFKA_BOOTSTRAP_SERVERS + # value: "kafka-cluster:9092" + + # Optional volume mounts for the SQL Gateway + volumeMounts: [] + # Example: + # volumeMounts: + # - name: kafka-certs + # mountPath: /opt/flink/certs + # readOnly: true + + # Optional volumes for the SQL Gateway + volumes: [] + # Example: + # volumes: + # - name: kafka-certs + # secret: + # secretName: kafka-cluster-ca-cert + +# SQL Client configuration (optional - for interactive SQL queries) +sqlClient: + enabled: false # Set to true if you need interactive SQL client + name: flink-sql-client + resources: + requests: + memory: 512Mi + cpu: 0.25 + limits: + memory: 1Gi + cpu: 0.5 + # Optional environment variables for the SQL Client + env: [] + # Example: + # env: + # - name: KAFKA_BOOTSTRAP_SERVERS + # value: "kafka-cluster:9092" + + # Optional volume mounts for the SQL Client + volumeMounts: [] + # Example: + # volumeMounts: + # - name: kafka-certs + # mountPath: /opt/flink/certs + # readOnly: true + + # Optional volumes for the SQL Client + volumes: [] + # Example: + # volumes: + # - name: kafka-certs + # secret: + # secretName: kafka-cluster-ca-cert + +# Service configuration +services: + sqlGateway: + type: ClusterIP + port: 8083 + targetPort: 8083 + sessionCluster: + type: ClusterIP + port: 8081 + targetPort: 8081 + +# RBAC configuration +rbac: + create: true + serviceAccountName: flink-sql-gateway + +# Security configuration +security: {} +# Example security configurations can be added here if needed +# security: +# runAsUser: 1000 +# runAsGroup: 1000 +# fsGroup: 1000 + +# Monitoring configuration +monitoring: + enabled: false + prometheus: + enabled: false + grafana: + enabled: false + +# Ingress configuration +ingress: + enabled: false + className: "" + annotations: {} + hosts: + - host: flink-sql-gateway.local + paths: + - path: / + pathType: Prefix + tls: [] + +# Node selector and tolerations +nodeSelector: {} +tolerations: [] +affinity: {} + +# Pod security context +podSecurityContext: {} + +# Container security context +securityContext: {} \ No newline at end of file