05. Spark 与端到端

目标：Spark 在 K8s 上跑批量，1M 行数据写 Iceberg 和 Doris，并由 DS 调度。

为什么 Spark 要跑在 K8s 上

driver/executor 按任务创建，跑完释放，不用长期占机器；
镜像自带 Iceberg/Kafka/JDBC 依赖，环境可复制；
DS 只需要创建 K8s Job，不在 worker 里跑计算；
代价：启动比 local 模式慢，日志和 UI 要额外做 SHS。

1. Spark 镜像

为什么把 jar 打进镜像：K8s 拉不到 --jars 的本地路径；镜像不可变，driver/executor 用同一个 tag，避免“driver 有 jar、executor 没有”。
为什么用 spark:3.5.9 而不是宿主机的 Spark：K8s 上每个 Pod 都要能独立启动，不能依赖 node1 的 /opt/spark。

1.1 Dockerfile

dockerfile

# 基础镜像：Spark 3.5.9 + Scala 2.12，自带 Hadoop/Python
FROM spark:3.5.9

# Iceberg Spark runtime，提供 SparkCatalog 与 Iceberg SQL 扩展
COPY jars/iceberg-spark-runtime-3.5_2.12-1.6.1.jar /opt/spark/jars/

# Iceberg 的 AWS S3 客户端，S3FileIO 需要
COPY jars/iceberg-aws-bundle-1.6.1.jar /opt/spark/jars/

# PostgreSQL JDBC 驱动，读 pg_orders / 写 JDBC
COPY jars/postgresql-42.7.4.jar /opt/spark/jars/

# MySQL JDBC 驱动，写 Doris（MySQL 协议）
COPY jars/mysql-connector-j-8.4.0.jar /opt/spark/jars/

# Kafka source/sink，需要时用
COPY jars/spark-sql-kafka-0-10_2.12-3.5.9.jar /opt/spark/jars/

# 把批量任务脚本放到镜像内，cluster mode 用 local:// 提交
COPY batch.py /opt/spark/batch.py

字段说明：

指令	作用
`FROM`	固定基础镜像，保证 driver/executor 版本一致
`COPY ... /opt/spark/jars/`	Spark 启动时自动加载 jar，无需 `--jars`
`COPY batch.py`	让 driver 能用 `local:///opt/spark/batch.py` 读取脚本

构建并推送：

bash

docker build -t <REGISTRY>/spark:3.5.9-iceberg .
docker push <REGISTRY>/spark:3.5.9-iceberg

1.2 ServiceAccount 与 RBAC

Spark driver 不是“跑完就结束”，它要向 K8s 创建/删除 executor Pod。没有 RBAC，driver 能启动但 executor 创建失败，表现为任务一直卡住。
测试期直接绑 edit，生产应按最小权限单独写 Role。

Spark driver/executor 需要创建 Pod：

bash

kubectl -n data create serviceaccount spark
kubectl create clusterrolebinding spark-edit \
  --clusterrole=edit \
  --serviceaccount=data:spark

验证：

bash

kubectl -n data get sa spark
kubectl get clusterrolebinding spark-edit

2. 批量任务

批量任务做两件事：先全表 count 验证能读，再 groupBy 写结果表。
count 不触发 shuffle，groupBy 会；如果 executor 在 shuffle 阶段崩，问题通常在序列化/向量化/内存，而不是“数据读不到”。

2.1 batch.py

python

from pyspark.sql import SparkSession

# appName 会显示在 SHS/Flink 之外的 Spark UI；K8s 里也作为 driver 名的一部分
spark = SparkSession.builder.appName("iceberg-batch-k8s").getOrCreate()

# 读 Iceberg 表；表名由 catalog.lake + namespace + table 组成
df = spark.table("lake.ods.pg_orders_uncompressed")

# count 只扫元数据/列统计，不触发 shuffle，用来确认表能读、行数对
print("ICEBERG_ROWS=", df.count())

# groupBy 会触发 shuffle，是真正验证 executor 计算/序列化的步骤
summary = df.groupBy("status").count()

# 写回 Iceberg；createOrReplace 会生成新的快照，不影响旧快照
summary.writeTo("lake.ods.pg_orders_summary") \
  .using("iceberg") \
  .createOrReplace()

# 打印完成标记，方便在 driver 日志里 grep
print("SUMMARY_WRITTEN")
spark.stop()

2.2 spark-submit

在 spark-client Pod 中执行：

bash

/opt/spark/bin/spark-submit \
  --master k8s://https://kubernetes.default.svc \
  --deploy-mode cluster \
  --name spark-ol-batch \
  --conf spark.kubernetes.container.image=<REGISTRY>/spark:3.5.9-iceberg \
  --conf spark.kubernetes.namespace=data \
  --conf spark.kubernetes.authenticate.serviceAccountName=spark \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  --conf spark.driver.memory=3g \
  --conf spark.executor.memory=3g \
  --conf spark.executor.instances=2 \
  --conf spark.sql.iceberg.vectorization.enabled=false \
  --conf spark.sql.parquet.enableVectorizedReader=false \
  --conf spark.eventLog.enabled=true \
  --conf spark.eventLog.dir=s3a://warehouse/spark-events \
  --conf spark.sql.catalog.lake=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.lake.type=jdbc \
  --conf spark.sql.catalog.lake.uri=jdbc:postgresql://postgres.data:5432/iceberg \
  --conf spark.sql.catalog.lake.jdbc.user=polaris \
  --conf spark.sql.catalog.lake.jdbc.password='<PG_PASS>' \
  --conf spark.sql.catalog.lake.warehouse=s3://warehouse/lake \
  --conf spark.sql.catalog.lake.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
  --conf spark.sql.catalog.lake.s3.endpoint=http://seaweedfs-s3.storage:8333 \
  --conf spark.sql.catalog.lake.s3.path-style-access=true \
  --conf spark.sql.catalog.lake.s3.access-key-id='<S3_AK>' \
  --conf spark.sql.catalog.lake.s3.secret-access-key='<S3_SK>' \
  --conf spark.sql.catalog.lake.client.region=us-east-1 \
  local:///opt/spark/batch.py

参数说明：

参数	作用
`--master k8s://...`	让 Spark 用 K8s 调度 driver/executor
`--deploy-mode cluster`	driver 跑在 K8s Pod，客户端退出不影响任务
`--name`	任务名，SHS/Job 里可见
`container.image`	driver/executor 都用这个镜像
`namespace`	driver/executor 创建在 `data`
`serviceAccountName`	创建 executor Pod 的身份
`driver.memory/executor.memory`	JVM 堆大小，按数据量调
`executor.instances`	初始 executor 数；没开动态分配就固定
`vectorization.enabled=false`	关闭 Iceberg 向量化，规避 exit 134
`parquet.enableVectorizedReader=false`	再关 Spark 自身的向量化
`eventLog.enabled/dir`	写事件日志给 SHS
`catalog.lake.*`	Iceberg catalog 与 S3 配置
`local://`	脚本在镜像内，不用上传

为什么关向量化：Iceberg 1.6.1 的向量化 Parquet reader 在这个 Spark 3.5 组合下会崩 executor，日志表现为 exit code 134。
代价是行式读取慢一些，但稳定；升级 Iceberg/Spark 后可以再压测决定是否打开。

验证：

bash

kubectl -n data get pods | grep spark-ol
kubectl -n data logs spark-ol-batch-...-driver | grep -E 'ICEBERG_ROWS|SUMMARY'

预期 ICEBERG_ROWS=1000012，driver Completed。

3. DS 的 K8S 任务

为什么不让 DS worker 直接 spark-submit：worker 里没有 Spark/kubeconfig，资源也不隔离；让 DS 创建一个 K8s Job，Job 里再提交 driver/executor，计算全在 K8s。
代价：Job 的 ServiceAccount 要有创建 Pod 的权限；任务参数必须是 K8S 插件要求的 JSON 结构。

3.1 参数

在 DS 里新建 K8S 任务，任务参数：

json

{
  "image": "<REGISTRY>/spark:3.5.9-iceberg",
  "namespace": "{\"name\":\"data\",\"cluster\":\"default\"}",
  "command": "[\"/opt/spark/bin/spark-submit\"]",
  "args": "[\"--master\",\"k8s://https://kubernetes.default.svc\",\"--deploy-mode\",\"cluster\",\"--conf\",\"spark.kubernetes.container.image=<REGISTRY>/spark:3.5.9-iceberg\",\"--conf\",\"spark.kubernetes.namespace=data\",\"--conf\",\"spark.kubernetes.authenticate.serviceAccountName=default\",\"--conf\",\"spark.executor.instances=2\",\"local:///opt/spark/batch.py\"]",
  "imagePullPolicy": "IfNotPresent",
  "minCpuCores": 2,
  "minMemorySpace": 8192
}

字段说明：

字段	作用
`image`	K8s Job 的镜像，必须含 spark-submit
`namespace`	DS K8s 集群/namespace，`name` 是 namespace，不是字段名
`command`	容器入口，JSON 数组
`args`	传给入口的参数，每个参数一个数组元素
`imagePullPolicy`	固定 tag 用 IfNotPresent，`:latest` 才会 Always
`minCpuCores`	Job 的 CPU 请求
`minMemorySpace`	Job 的内存请求，单位 Mi；8192=8Gi

3.2 DS 侧注册 K8s 集群

sql

-- t_ds_cluster.config 必须是 JSON
-- {"k8s":"<admin.conf 的 YAML 内容>"}

namespace 的 key 是 name，不是 namespace：

json

{"name":"data","cluster":"default"}

3.3 常见错误

报错	原因
`k8sTaskExecutionContext` NPE	`t_ds_cluster` 没注册或 `cluster` 名字不对
Job 名 invalid	任务名有下划线，必须 RFC1123
`OOMKilled`	`minMemorySpace` 单位是 Mi，8 会变成 8Mi
`ClassCastException`	`command/args` 不是 JSON 数组字符串

4. 1M 数据

为什么要造 1M 行：小数据量看不出 shuffle、Iceberg 文件数、Doris 写入瓶颈；1M 是能在 3 节点上跑完、又能暴露问题的量级。
为什么固定 seed：每次重建数据一致，才能对比“这次比上次快/慢”。

4.1 MySQL 补数

MySQL 没有 generate_series，用 Python 分批生成 SQL，每批 5 万行，避免 max_allowed_packet：

python

# gen_mysql_1m.py
start, end, batch = 501013, 1000012, 50000
idx = 0
for s in range(start, end + 1, batch):
    e = min(s + batch - 1, end)
    with open(f"/tmp/mysql_1m_{idx}.sql", "w") as f:
        f.write("INSERT IGNORE INTO orders (id,user_id,amount,status) VALUES\n")
        rows = [
            f"({i},{3000+i},{round(10+i*0.37,2)},'{(['NEW','PAID','NEW','PAID'][i%4])}')"
            for i in range(s, e + 1)
        ]
        f.write(",\n".join(rows) + ";\n")
    idx += 1

执行：

bash

python3 gen_mysql_1m.py
for f in /tmp/mysql_1m_*.sql; do
  kubectl -n data exec -i mysql-0 -- \
    mysql -uroot -p"$MYSQL_PASS" demo < "$f"
done

验证：

bash

kubectl -n data exec -i mysql-0 -- \
  mysql -uroot -p"$MYSQL_PASS" -N -e 'SELECT count(*) FROM demo.orders;'

4.2 PostgreSQL 补数

sql

INSERT INTO pg_orders (id, user_id, amount, status)
SELECT g, 4000+g, round((10+g*0.41)::numeric,2),
       (ARRAY['NEW','PAID','NEW','PAID'])[(g%4)+1]
FROM generate_series(501013,1000012) g
ON CONFLICT (id) DO NOTHING;

验证：

sql

SELECT count(*) FROM demo.orders;   -- 1000012
SELECT count(*) FROM pg_orders;     -- 1000012

4.3 生成 uncompressed Iceberg 表

如果不想读旧 zstd 文件，直接从 PG 重新生成：

python

df = (spark.read.format("jdbc")
      .option("url", "jdbc:postgresql://postgres.data:5432/demo")
      .option("dbtable", "public.pg_orders")
      .option("user", "polaris")
      .option("password", "<PG_PASS>")
      .load())

(df.writeTo("lake.ods.pg_orders_uncompressed")
 .using("iceberg")
 .tableProperty("write.parquet.compression-codec", "uncompressed")
 .createOrReplace())

5. 写 Doris：Stream Load

为什么不用 JDBC 逐批 INSERT：1M 行会产生大量小事务，Doris 要反复建事务、写 tablet，实测极慢；Stream Load 一次传 CSV，BE 直接批量导入。
代价：要先落地 CSV，且列顺序/分隔符必须和目标表一致。

不要用 JDBC 逐批写 1M，直接用 Stream Load。

5.1 导出 CSV

python

df = spark.table("lake.ods.pg_orders_uncompressed") \
  .select("id", "user_id", "amount", "status", "updated_at") \
  .coalesce(1)

df.write.mode("overwrite").option("header", "false").csv("/tmp/doris_export")

5.2 清空目标表

bash

kubectl -n doris exec doris-fe-0 -- mysql -h127.0.0.1 -P9030 -uroot -e \
  "TRUNCATE TABLE demo.iceberg_orders;"

5.3 执行 Stream Load

bash

FILE=$(ls /tmp/doris_export/part-* | head -1)

curl --location-trusted -u root: \
  -H "label: iceberg_orders_$(date +%s)" \
  -H "format: csv" \
  -H "column_separator:," \
  -H "Expect:100-continue" \
  -T "$FILE" \
  http://doris-fe-service.doris.svc.cluster.local:8030/api/demo/iceberg_orders/_stream_load

参数说明：

参数/字段	作用
`FILE`	Spark `coalesce(1)` 输出的唯一 part 文件
`-u root:`	Doris 用户；测试 root 空密码，生产用专用导入账号
`label`	幂等标签，重复提交同 label 会被拒绝，防止重复导入
`format: csv`	按 CSV 解析
`column_separator: ,`	列分隔符，必须和导出格式一致
`Expect:100-continue`	大文件先探活，避免传一半才失败
`-T`	上传整个文件
URL `/api/{db}/{table}/_stream_load`	Doris Stream Load 固定端点

预期：

json

{
  "Status": "Success",
  "NumberLoadedRows": 1000012,
  "NumberFilteredRows": 0,
  "LoadTimeMs": 502
}

5.4 验证

bash

kubectl -n doris exec doris-fe-0 -- mysql -h127.0.0.1 -P9030 -uroot -N -e \
  "SELECT count(*) FROM demo.iceberg_orders;"

预期 1000012。

6. 本阶段验收

bash

kubectl -n data get pods | grep spark
kubectl -n doris exec doris-fe-0 -- mysql -h127.0.0.1 -P9030 -uroot -N \
  -e "SELECT count(*) FROM demo.iceberg_orders;"

要求：

Spark driver/executor Completed；
Iceberg 1,000,012 行；
Doris 1,000,012 行；
SHS 能查到 iceberg-batch-k8s。

下一步：排障与清理。

01.Mybatis

02.Spring

03.SpringMVC

04.SSM

05.MybatisPlus

06.SpringBoot

07.SpringBootWeb

08.Swagger

09.SpringCloud

10.SpringCloudAlibaba

11.Others

01.基础

02.爬虫

03. 选型决策

04. 部署实施

05. 实战与验收

5.1 实战教程

05. Spark 与端到端

为什么 Spark 要跑在 K8s 上

1. Spark 镜像

1.1 Dockerfile

1.2 ServiceAccount 与 RBAC

2. 批量任务

2.1 batch.py

2.2 spark-submit

3. DS 的 K8S 任务

3.1 参数

3.2 DS 侧注册 K8s 集群

3.3 常见错误

4. 1M 数据

4.1 MySQL 补数

4.2 PostgreSQL 补数

4.3 生成 uncompressed Iceberg 表

5. 写 Doris：Stream Load

5.1 导出 CSV

5.2 清空目标表

5.3 执行 Stream Load

5.4 验证

6. 本阶段验收

5.1 实战教程

05. Spark 与端到端 ​

为什么 Spark 要跑在 K8s 上 ​

1. Spark 镜像 ​

1.1 Dockerfile ​

1.2 ServiceAccount 与 RBAC ​

2. 批量任务 ​

2.1 batch.py ​

2.2 spark-submit ​

3. DS 的 K8S 任务 ​

3.1 参数 ​

3.2 DS 侧注册 K8s 集群 ​

3.3 常见错误 ​

4. 1M 数据 ​

4.1 MySQL 补数 ​

4.2 PostgreSQL 补数 ​

4.3 生成 uncompressed Iceberg 表 ​

5. 写 Doris：Stream Load ​

5.1 导出 CSV ​

5.2 清空目标表 ​

5.3 执行 Stream Load ​

5.4 验证 ​

6. 本阶段验收 ​

05. Spark 与端到端

为什么 Spark 要跑在 K8s 上

1. Spark 镜像

1.1 Dockerfile

1.2 ServiceAccount 与 RBAC

2. 批量任务

2.1 batch.py

2.2 spark-submit

3. DS 的 K8S 任务

3.1 参数

3.2 DS 侧注册 K8s 集群

3.3 常见错误

4. 1M 数据

4.1 MySQL 补数

4.2 PostgreSQL 补数

4.3 生成 uncompressed Iceberg 表

5. 写 Doris：Stream Load

5.1 导出 CSV

5.2 清空目标表

5.3 执行 Stream Load

5.4 验证

6. 本阶段验收