69 lines
2.7 KiB
YAML
69 lines
2.7 KiB
YAML
apiVersion: sparkoperator.k8s.io/v1beta2
|
|
kind: SparkApplication
|
|
metadata:
|
|
name: cdc-{{ .Values.customer }}
|
|
namespace: {{ .Release.Namespace }}
|
|
spec:
|
|
type: Python
|
|
mode: cluster
|
|
image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
|
|
imagePullPolicy: {{ .Values.image.pullPolicy }}
|
|
imagePullSecrets:
|
|
- {{ .Values.imagePullSecret }}
|
|
mainApplicationFile: "local:///opt/spark/work-dir/cdc_crm_to_raw.py"
|
|
sparkVersion: "{{ .Values.spark.version }}"
|
|
# streaming job chạy mãi -> restart nếu chết (thay cho việc submit tay foreground)
|
|
restartPolicy:
|
|
type: Always
|
|
onFailureRetryInterval: 30
|
|
onSubmissionFailureRetryInterval: 30
|
|
sparkConf:
|
|
# Iceberg catalog (Polaris REST) + MinIO
|
|
"spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
|
|
"spark.sql.catalog.iceberg": "org.apache.iceberg.spark.SparkCatalog"
|
|
"spark.sql.catalog.iceberg.type": "rest"
|
|
"spark.sql.catalog.iceberg.uri": "{{ .Values.catalog.uri }}"
|
|
"spark.sql.catalog.iceberg.warehouse": "{{ .Values.catalog.warehouse }}"
|
|
"spark.sql.catalog.iceberg.header.X-Iceberg-Access-Delegation": "vended-credentials"
|
|
"spark.sql.catalog.iceberg.io-impl": "org.apache.iceberg.aws.s3.S3FileIO"
|
|
"spark.sql.catalog.iceberg.s3.endpoint": "{{ .Values.s3.endpoint }}"
|
|
"spark.sql.catalog.iceberg.s3.path-style-access": "true"
|
|
"spark.sql.catalog.iceberg.s3.region": "{{ .Values.s3.region }}"
|
|
"spark.sql.defaultCatalog": "iceberg"
|
|
driver:
|
|
cores: {{ .Values.spark.driverCores }}
|
|
memory: "{{ .Values.spark.driverMemory }}"
|
|
labels: { app: cdc-{{ .Values.customer }} }
|
|
env:
|
|
- name: KAFKA_BOOTSTRAP
|
|
value: "{{ .Values.kafka.bootstrap }}"
|
|
- name: TOPIC_PATTERN
|
|
value: "{{ .Values.topicPattern }}"
|
|
- name: TARGET_TABLE
|
|
value: "{{ .Values.targetTable }}"
|
|
- name: TRIGGER_INTERVAL
|
|
value: "{{ .Values.triggerInterval }}"
|
|
- name: MAX_OFFSETS_PER_TRIGGER
|
|
value: "{{ .Values.maxOffsetsPerTrigger }}"
|
|
- name: CHECKPOINT
|
|
value: "file:///checkpoints/cdc_{{ .Values.customer }}"
|
|
volumeMounts:
|
|
- name: checkpoint
|
|
mountPath: /checkpoints
|
|
executor:
|
|
instances: {{ .Values.spark.executorInstances }}
|
|
cores: {{ .Values.spark.executorCores }}
|
|
memory: "{{ .Values.spark.executorMemory }}"
|
|
volumeMounts:
|
|
- name: checkpoint
|
|
mountPath: /checkpoints
|
|
volumes:
|
|
- name: checkpoint
|
|
persistentVolumeClaim:
|
|
claimName: cdc-{{ .Values.customer }}-checkpoint
|
|
deps:
|
|
packages:
|
|
- org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.9.1
|
|
- org.apache.iceberg:iceberg-aws-bundle:1.9.1
|
|
- org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.6
|