当前位置：首页 > news >正文

Kubernetes大数据处理实践

news 2026/5/11 2:59:28

Kubernetes大数据处理实践

一、引言

大数据处理是现代企业的核心需求之一。Kubernetes为大数据处理提供了弹性、可扩展的平台支持，能够高效运行Spark、Flink等大数据框架。

二、大数据处理架构

2.1 大数据处理参考架构

┌─────────────────────────────────────────────────────────────────┐ │ 大数据处理架构 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 数据源 │───▶│ 数据存储 │───▶│ 计算引擎 │───▶│ 结果输出 │ │ │ │ (Kafka) │ │ (HDFS/S3) │ │ (Spark) │ │ (DB/DW) │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ ▼ │ │ ┌──────────┐ │ │ │ 资源管理 │ │ │ │ (YARN/K8s) │ │ │ └──────────┘ │ └─────────────────────────────────────────────────────────────────┘

2.2 大数据框架对比

框架	类型	适用场景
Apache Spark	批处理/流处理	通用大数据处理
Apache Flink	流处理	实时流处理
Apache Kafka Streams	流处理	轻量级流处理
Apache Hadoop	批处理	传统大数据批处理

三、Spark on Kubernetes部署

3.1 Spark Operator部署

# 安装Spark Operator kubectl apply -f https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/releases/download/v1.1.0/spark-operator.yaml # 查看Operator状态 kubectl get pods -n spark-operator

3.2 Spark Application配置

apiVersion: sparkoperator.k8s.io/v1beta2 kind: SparkApplication metadata: name: spark-pi namespace: default spec: type: Scala mode: cluster image: gcr.io/spark-operator/spark:v3.4.1 imagePullPolicy: Always mainClass: org.apache.spark.examples.SparkPi mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.4.1.jar sparkVersion: "3.4.1" restartPolicy: type: OnFailure onFailureRetries: 3 onFailureRetryInterval: 10 onSubmissionFailureRetries: 5 onSubmissionFailureRetryInterval: 20 driver: cores: 1 coreLimit: "1200m" memory: "512m" labels: version: 3.4.1 serviceAccount: spark executor: cores: 1 instances: 3 memory: "1024m" labels: version: 3.4.1

3.3 Spark ServiceAccount配置

apiVersion: v1 kind: ServiceAccount metadata: name: spark namespace: default --- apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: spark-role namespace: default rules: - apiGroups: [""] resources: ["pods", "services", "configmaps"] verbs: ["*"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding metadata: name: spark-role-binding namespace: default roleRef: apiGroup: rbac.authorization.k8s.io kind: Role name: spark-role subjects: - kind: ServiceAccount name: spark namespace: default

四、Flink on Kubernetes部署

4.1 Flink Deployment配置

apiVersion: flink.apache.org/v1beta1 kind: FlinkDeployment metadata: name: flink-cluster namespace: default spec: image: flink:1.17.1 flinkVersion: v1_17 flinkConfiguration: taskmanager.numberOfTaskSlots: "4" jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m serviceAccount: flink jobManager: replicas: 1 resource: memory: "2048m" cpu: 1 taskManager: replicas: 2 resource: memory: "4096m" cpu: 2

4.2 Flink Job配置

apiVersion: flink.apache.org/v1beta1 kind: FlinkSessionJob metadata: name: my-flink-job namespace: default spec: deploymentName: flink-cluster job: jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar parallelism: 4 upgradeMode: stateless

4.3 Flink ServiceAccount配置

apiVersion: v1 kind: ServiceAccount metadata: name: flink namespace: default --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: flink rules: - apiGroups: [""] resources: ["pods", "services", "configmaps", "events"] verbs: ["*"] - apiGroups: ["apps"] resources: ["deployments"] verbs: ["*"] - apiGroups: ["flink.apache.org"] resources: ["flinkdeployments", "flinksessionjobs"] verbs: ["*"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: name: flink roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole name: flink subjects: - kind: ServiceAccount name: flink namespace: default

五、Kafka on Kubernetes部署

5.1 Kafka集群配置

apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster namespace: kafka spec: kafka: version: 3.5.1 replicas: 3 listeners: - name: plain port: 9092 type: internal tls: false - name: tls port: 9093 type: internal tls: true config: offsets.topic.replication.factor: 3 transaction.state.log.replication.factor: 3 transaction.state.log.min.isr: 2 storage: type: jbod volumes: - id: 0 type: persistent-claim size: 100Gi deleteClaim: false zookeeper: replicas: 3 storage: type: persistent-claim size: 50Gi deleteClaim: false entityOperator: topicOperator: {} userOperator: {}

5.2 Kafka Topic配置

apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaTopic metadata: name:># 安装Airflow helm install airflow apache-airflow/airflow --namespace airflow # 查看Airflow状态 kubectl get pods -n airflow

6.2 Airflow DAG配置

from datetime import datetime, timedelta from airflow import DAG from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2024, 1, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'spark_data_processing', default_args=default_args, description='Spark data processing pipeline', schedule_interval=timedelta(days=1), ) spark_task = SparkSubmitOperator( task_id='spark_job', application='gs://bucket/spark-job.jar', name='data-processing', conf={ 'spark.executor.instances': '5', 'spark.executor.memory': '2g', 'spark.driver.memory': '1g' }, dag=dag, )

七、大数据监控与可观测性

7.1 Spark指标收集

apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: spark-monitor spec: selector: matchLabels: app: spark endpoints: - port: metrics interval: 30s scrapeTimeout: 10s

7.2 Prometheus查询

# Spark作业执行时间 sum(rate(spark_job_executor_run_time_total[5m])) by (job_id) # Spark任务完成数 sum(spark_job_tasks_completed) by (job_id) # Flink作业延迟 avg(flink_taskmanager_job_task_operator_latency_max) by (job_name)

7.3 Grafana仪表盘

{ "title": "Big Data Processing Metrics", "panels": [ { "type": "graph", "targets": [ { "expr": "sum(rate(spark_job_executor_run_time_total[5m])) by (job_id)", "legendFormat": "{{job_id}}" } ] }, { "type": "stat", "targets": [ { "expr": "sum(spark_job_tasks_completed)", "legendFormat": "Completed Tasks" } ] } ] }