前阵子收尾一套离线数仓的链路改造，核心工作只有一句话：用 Scala 和 Spark 从 HDFS 里把数据算完，然后稳定地写进 ClickHouse，再让报表侧从 ClickHouse 里把聚合结果读出来。听起来是件很平常的事，但真正落地的时候，麻烦全藏在细节里——驱动包路径换了、分区字段选错了、写入批次太小把 ClickHouse 的 part 数量顶爆、时区差八小时、Decimal 精度对不上。这篇就按实际动手的顺序，把 Hadoop 生态里 Spark 与 ClickHouse 的对接完整走一遍，从选型、依赖、读、写、调参到排查，能给代码的地方给代码，能给算式的地方给算式。适合手里有 Spark 批处理任务、需要把结果落到 ClickHouse 做 OLAP 查询的同学，也适合刚接手这块链路、被各种连接异常折腾过的同行。
1. 这套组合到底解决什么问题：选型与整体设计
先说清楚链路的样子，不然后面的参数取舍没有依据。典型场景是：原始日志或业务库的增量数据落在 HDFS/Hive 上，Spark 用 Scala 写批处理作业，做清洗、关联、聚合，产出一张宽表或聚合表，最后写进 ClickHouse；下游的报表、看板、即席查询直接打 ClickHouse。反向也常见，ClickHouse 里存着明细或维表，Spark 需要把它拉回来跟 HDFS 上的数据做关联，这时候就是读。
为什么不让 Spark 直接算完写回 Hive，报表查 Hive？因为交互式查询的响应差得太远。Hive 走 MapReduce 或 Tez，一条多维度聚合查询十几秒起步；ClickHouse 在同样数据量上通常能压到几百毫秒到一两秒。反过来，为什么不让 ClickHouse 直接吃原始数据做全量计算？因为 ClickHouse 擅长的是单表扫描加聚合，多表大关联、复杂 UDF、迭代计算这类活儿它做得很吃力，而且一旦把重计算压到 CK 上，会跟报表查询抢资源。所以合理的分工是：Spark 负责重计算，ClickHouse 负责高性能的查询服务，边界清晰。
1.1 Spark 与 ClickHouse 在链路里的职责边界
我把这条链路的职责划成三段。第一段是数据接入，这一层完全交给 HDFS，按天分区落地原始数据，保证可回溯、可重跑。第二段是计算，Spark 承担全部的重活，包括去重、关联维表、窗口计算、指标汇总，输出的是已经按 ClickHouse 表结构组织好的结果集。第三段是服务，ClickHouse 只做一件事：把结果表按查询模式建好索引和排序键，然后对外提供低延迟查询。
这个边界定下来之后，有个很自然的问题：Spark 的计算结果要不要在 HDFS 上再存一份？我的建议是存。ClickHouse 的表虽然也有副本，但它的定位是查询服务层，不适合当作唯一数据源。一旦某次写入逻辑写错、或者建表语句改错，没有 HDFS 上的中间结果就只能全链路重跑，代价可能是几个小时。多存一份 Parquet，成本是磁盘，换来的是可重放能力，这笔账很划算。
还有一点值得提前说：不要试图让 Spark 承担 ClickHouse 的部分查询职责，反过来也不要。我见过有团队为了"统一入口"，在 ClickHouse 前面套一层 Spark ThriftServer 转发查询，结果是延迟翻了几倍，还多了一个不稳定组件。工具各司其职，链路才稳。
1.2 为什么用 Scala 而不是 PySpark 或 Java
语言选择这件事在批处理场景里其实挺关键。PySpark 写起来快，但有两个绕不开的成本：一是 JVM 与 Python 进程之间的序列化开销，在 UDF 密集的作业里能占到总耗时的两三成；二是依赖管理，Python 侧要额外处理解释器路径、第三方包分发，集群上 pyenv 和环境隔离一旦出问题，排查成本很高。Java 的问题相反，冗长，写一个 DataSet 转换要一堆样板代码，迭代效率低。
Scala 是 Spark 的原生语言，API 最完整，Dataset[T]的强类型转换、Encoder自动派生、模式匹配处理脏数据，写起来都很顺手。特别是处理 ClickHouse 返回的Nullable字段时，Scala 的Option配合getAs[Option[T]]能让空值处理显式化，比 Java 里到处判 null 干净得多。代价是团队需要一定的 Scala 基础，但这个门槛在写两三个作业之后基本就跨过去了。
需要提醒的是 Scala 版本必须跟 Spark 版本对齐，这个坑后面单独讲。Spark 3.3 之后默认 Scala 2.12，Spark 3.5 开始提供 2.13 的构建，混用会直接抛NoSuchMethodError，报错位置还在很深的调用栈里，第一次遇到很难定位。
1.3 三种连接方式的取舍
从 Spark 连 ClickHouse，市面上主要三条路，我做了个对照表：
| 方式 | 依赖 | 读写支持 | 并行度控制 | 适用场景 | 主要短板 |
|---|---|---|---|---|---|
| 原生 JDBC（Spark 内置 jdbc format） | clickhouse-jdbc 驱动 | 读写都支持 | 读靠 partitionColumn，写靠 repartition | 结构简单、字段类型规整的表 | 写入逐行 addBatch，批次控制粗糙 |
| ClickHouse 官方 Spark 连接器 | clickhouse-spark-connector | 读写都支持 | 内置分片感知，可分片写入本地表 | 大规模写入、需要写本地表 | 版本耦合紧，配置项多 |
| HTTP 接口 + 自定义 Sink | 仅需 HTTP 客户端 | 主要做写 | 完全自己控制 | 有特殊协议或鉴权需求的场景 | 全部手写，维护成本高 |
绝大多数场景我推荐先用原生 JDBC 把流程跑通，理由是依赖最少、行为最可预期、出问题好定位。等链路稳定了、写入量上来了，再评估是否换成官方连接器。官方连接器的优势在于它理解 ClickHouse 的分布式表结构，能把数据直接分发到各个分片的本地表，避免所有写入都压到分布式表的入口节点上。但如果你的集群规模不大、写入量在每秒几万行以内，JDBC 完全够用，没必要为了"更专业"引入额外复杂度。
自定义 HTTP Sink 我只在一种情况下用过：需要对写入做极致的攒批控制，或者要在写入前后插入自定义的校验逻辑。这条路能把性能榨到最高，但每个细节都得自己实现，包括失败重试、幂等保证、连接池管理，工作量不小。
2. 环境与依赖准备：版本对齐是第一道坎
依赖这一步看起来最没技术含量，实际上是我见过最多人卡住的地方。ClickHouse 的 JDBC 驱动在 0.3.x 到 0.6.x 之间做过一次彻底的包名重构，驱动类从ru.yandex.clickhouse.ClickHouseDriver变成了com.clickhouse.jdbc.ClickHouseDriver，老代码直接升级依赖会报ClassNotFoundException。这不是配置问题，是包路径真的变了。
2.1 版本兼容矩阵与必须核对的四项
在动手写代码前，把这四样东西的版本列出来对齐：Spark 版本、Scala 二进制版本、JDK 版本、clickhouse-jdbc 版本。它们之间是连锁关系，错一个就编译不过或者运行时报错。
| 组件 | 推荐组合 | 说明 |
|---|---|---|
| Spark | 3.3.x / 3.4.x / 3.5.x | 3.3 起 Scala 默认 2.12 |
| Scala | 2.12（Spark 3.5 可选 2.13） | 必须与 Spark 构建版本一致 |
| JDK | 8 或 11 | Spark 3.3+ 建议 11，CK 驱动都兼容 |
| clickhouse-jdbc | 0.4.6 / 0.6.x | 0.6 起包名为 com.clickhouse.jdbc |
注意：
clickhouse-jdbc从 0.6 开始拆成了clickhouse-jdbc（纯 HTTP）和clickhouse-jdbc-all（含 native 协议）两个 artifact，同时底层依赖clickhouse-http-client。如果只引clickhouse-jdbc又想走 9000 端口的 native 协议，会缺类。我一般直接引clickhouse-jdbc的 shaded 版本，省事。
JDK 版本这块还有个细节：Spark 3.3 在 JDK 17 上跑需要额外加一堆--add-opens参数，否则IllegalAccessError到处都是。生产环境如果没特殊要求，JDK 11 是最省心的选择，兼容性好，也不用折腾模块系统。
2.2 Maven 依赖清单与打包注意事项
依赖按下面这样引，版本号统一放在properties里便于升级：
<properties> <spark.version>3.4.2</spark.version> <scala.binary.version>2.12</scala.binary.version> <clickhouse.jdbc.version>0.4.6</clickhouse.jdbc.version> </properties> <dependencies> <!-- Spark 相关，scope 设为 provided，集群上已有 --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> </dependency> <!-- ClickHouse 驱动，必须打进 fat jar --> <dependency> <groupId>com.clickhouse</groupId> <artifactId>clickhouse-jdbc</artifactId> <version>${clickhouse.jdbc.version}</version> <classifier>all</classifier> </dependency> </dependencies>这里有个常见的翻车点：Spark 相关依赖设成provided是对的，但 ClickHouse 驱动千万别也设成provided，否则提交到 YARN 上每个 executor 都找不到驱动类，报No suitable driver。我第一次踩这个坑时本地跑得好好的，一上集群就挂，排查了半天才发现是 scope 写错了。
打包用maven-shade-plugin，注意要把META-INF/services/java.sql.Driver做合并，否则多个驱动（比如同时有 MySQL 驱动）的 SPI 文件会互相覆盖，导致DriverManager找不到 ClickHouse 驱动。shade 插件里加ServicesResourceTransformer就能解决。
2.3 ClickHouse 侧的建表与账号权限准备
代码写之前先把 ClickHouse 这边准备好。建一张用于写入的结果表，引擎选ReplacingMergeTree，排序键按查询模式定：
CREATE TABLE analytics.dws_order_daily ( stat_date Date, city LowCardinality(String), order_id UInt64, amount Decimal(18, 2), update_time DateTime ) ENGINE = ReplacingMergeTree(update_time) PARTITION BY toYYYYMM(stat_date) ORDER BY (stat_date, city, order_id) SETTINGS index_granularity = 8192;排序键的顺序很讲究，要按"查询里出现频率高、过滤性强的字段在前"来排。上面这个表如果最常见的查询是"按日期范围 + 城市过滤"，那(stat_date, city, order_id)就是合理的；如果反过来把order_id放第一位，日期范围查询就用不上索引，扫描量会大很多。
账号权限要单独建一个写入用户，不要用 default。最小权限原则在这里同样适用：
CREATE USER spark_rw IDENTIFIED WITH sha256_password BY '***'; GRANT SELECT, INSERT ON analytics.* TO spark_rw; -- 如果需要读系统表做监控 GRANT SELECT ON system.query_log TO spark_rw;提示：ClickHouse 的连接数上限由
max_concurrent_queries控制（默认 100），Spark 侧每个 task 开一个连接，如果分区数开到 200，写入时会有一半连接被拒，报TOO_MANY_SIMULTANEOUS_QUERIES。这个参数要么在 CK 侧调大，要么在 Spark 侧用repartition把并行度压下来。我一般倾向后者，因为 CK 侧的资源也是有限的。
3. Spark 读 ClickHouse：分区并行拉取的实现细节
读的场景通常有两个：一是全量初始化，把 ClickHouse 里的维表或历史数据拉到 Spark 做一次性回刷；二是增量关联，每天拉当天的明细去跟 HDFS 数据做 join。两种场景对并行度的要求不一样，全量初始化希望把机器跑满，增量关联希望别给 CK 太大压力。
3.1 JDBC 读取的基础写法与参数含义
先给一版能直接跑的读取代码：
val spark = SparkSession.builder() .appName("ck-read-demo") .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .getOrCreate() val jdbcUrl = "jdbc:clickhouse://ck-node-01:8123/analytics" + "?socket_timeout=600000&connect_timeout=30000&compress=1" val df = spark.read .format("jdbc") .option("url", jdbcUrl) .option("dbtable", "(SELECT order_id, stat_date, city, amount FROM analytics.dws_order_daily WHERE stat_date >= '2024-01-01') t") .option("user", "spark_rw") .option("password", "***") .option("driver", "com.clickhouse.jdbc.ClickHouseDriver") .option("partitionColumn", "order_id") .option("numPartitions", "16") .option("fetchsize", "10000") .load()几处容易被忽略的地方。dbtable如果写成子查询，必须用括号包起来并且给别名，t这个别名不能省，否则 Spark 拼出来的 SQL 语法不合法。url里加compress=1能显著降低网络传输量，对于宽表读取收益很明显，实测压缩比通常在 5:1 以上。fetchsize控制单次从游标里取多少行，默认值偏小，读大表时频繁往返会拖慢速度，调到 10000 左右比较合适，但也不要无限大，否则 executor 内存压力上升。
还有一个隐藏成本：如果不指定lowerBound和upperBound，Spark 会先对partitionColumn跑一次SELECT MIN(col), MAX(col)来算边界。这一步在 ClickHouse 上是全表扫描，如果partitionColumn不是排序键的前缀字段，这一次查询可能就要几十秒。所以要么显式给出上下界，要么保证分区字段在排序键里。
3.2 分片字段的选择逻辑与倾斜计算
partitionColumn的选择直接决定读取是否均衡。硬性要求只有一个：必须是数值类型（整型、浮点、Date 在部分版本也能用但行为不稳）。在这个前提下，选哪个字段有讲究。
最理想的是自增主键或订单号这类分布均匀的数值列。最糟糕的是有明显倾斜的字段，比如city_id，如果某个城市占了全部数据的一半，那不管切成多少份，总有一份特别大。可以用下面这段 SQL 先摸一下分布：
SELECT city_id, count() AS cnt FROM analytics.dws_order_daily GROUP BY city_id ORDER BY cnt DESC LIMIT 10;如果发现第一名的占比超过 20%，就别拿它当分区字段，换个更均匀的。实在没有合适的数值列，可以拿字符串主键的哈希值：cityHash64(order_id)或者toUInt64(sipHash64(order_id))，只要在dbtable的子查询里把它算成别名，Spark 就能用。
分区数的估算有个粗略公式：numPartitions ≈ 期望总并发 ÷ 单查询可接受的数据量。假设表里 2 亿行、单行 200 字节、总共约 40GB，每个分区控制在 1.5GB 到 2.5GB 之间比较合适，那 16 到 24 个分区是合理的。分区太多，每个分区数据量太小，调度开销和连接开销反而成为瓶颈；分区太少，单个 task 处理几 GB，容易 OOM 不说，CK 侧一个查询就要扫很久。
还有一个反直觉的情况：如果分区字段的值集中在很窄的区间，Spark 按范围切分会切出一堆空分区。比如order_id是从 1 到 100 万连续，但实际数据只有 90 万到 100 万这一段，切成 16 份的话前 14 份全是空的，真正干活的只有 2 个 task。这种情况要么手动指定lowerBound和upperBound贴合真实范围，要么改用哈希函数做分区字段。
3.3 类型映射与空值处理的实测记录
ClickHouse 和 Spark 的类型系统不是一一对应的，读的时候有些地方会出乎意料：
| ClickHouse 类型 | Spark 读出的类型 | 需要注意的点 |
|---|---|---|
| UInt64 / Int64 | LongType | UInt64 超出 Long 上限时会被截断或报错 |
| UInt32 | LongType | 会做向上兼容转换 |
| Decimal(P, S) | DecimalType(P, S) | 精度一致，但 Spark 侧做运算要注意 scale |
| Date / DateTime | DateType / TimestampType | 时区问题见下文 |
| LowCardinality(String) | StringType | 正常映射，性能更好 |
| Nullable(T) | T（允许 null） | 读取时 null 会正常保留 |
| Array(T) | ArrayType | JDBC 方式读取可能失败，建议用官方连接器 |
| FixedString(N) | StringType | 会补零，需要trim处理 |
最麻烦的是时区。ClickHouse 的DateTime不带时区信息，存储的是按列或服务端时区折算后的值。JDBC 驱动在转换时又会按 JVM 默认时区处理，两边一叠加就出现固定偏移。解决办法是在连接串上明确指定：use_time_zone=UTC&use_server_time_zone=false，让驱动不做二次转换，同时在 Spark 侧统一用 UTC 处理时间，展示层再转本地时区。这一条我踩过两次，前后端对着时间戳排查了一下午，最后发现就差八小时。
UInt64这个类型也要留意。ClickHouse 的UInt64最大值约 1.8e19，而 Spark 的LongType上限是 9.2e18，如果业务上有雪花 ID 这类可能超过 Long 上限的值，读出来会变成负数。稳妥的做法是在 SQL 里转成字符串读，或者确认业务 ID 不会触顶。
4. Spark 写 ClickHouse：三种落地姿势与攒批策略
写入是这条链路上最容易出问题、也最值得花时间打磨的部分。核心矛盾只有一个：Spark 的并行模型是"多个 task 各写各的"，ClickHouse 的 MergeTree 引擎则希望"每次写入形成尽量少、尽量大的 part"。task 数量一多、批次一小，part 数量就会指数级增长，最终撞上TOO_MANY_PARTS这个经典报错。
4.1 用 jdbc format 直接 append 的写法与适用边界
最省事的写法是先repartition再用内置的 jdbc format：
val resultDf = spark.sql("...") // 已按目标表结构组织好 resultDf .repartition(4) .write .mode("append") .format("jdbc") .option("url", jdbcUrl) .option("dbtable", "analytics.dws_order_daily") .option("user", "spark_rw") .option("password", "***") .option("driver", "com.clickhouse.jdbc.ClickHouseDriver") .option("batchsize", "50000") .option("isolationLevel", "NONE") .option("truncate", "false") .save()repartition(4)是这里最关键的一行。如果不写，分区数等于上游的 shuffle 分区数或者文件数，可能是 200 甚至更多，每个 task 建一个连接、各写一个小批次，ClickHouse 侧瞬间收到几百个小 part。isolationLevel设成NONE可以减少事务相关开销，ClickHouse 本身也不支持真正的事务隔离级别，设成其它值没意义还可能报错。
这种写法的适用边界很明确：单次写入量在几百万行以内、任务不频繁（比如每天几次）、可以接受把并行度压到很低。一旦超出这个范围，比如单次要写几千万行，把并行度压到 4 意味着单个 task 要处理上千万行，耗时会很难看，而且 CK 侧单查询的内存压力也上来了。这时候就得换第二种写法。
4.2 foreachPartition 手工攒批：控制力最强的方案
这是我在生产环境用的方案。核心思路是放弃 Spark 内置的写入器，自己在foreachPartition里管理连接和批次，把攒批逻辑完全握在手里：
val INSERT_SQL = "INSERT INTO analytics.dws_order_daily (stat_date, city, order_id, amount, update_time) VALUES (?, ?, ?, ?, ?)" val BATCH_SIZE = 50000 resultDf.repartition(8).foreachPartition { rows => var conn: java.sql.Connection = null var stmt: java.sql.PreparedStatement = null try { conn = java.sql.DriverManager.getConnection(jdbcUrl, "spark_rw", "***") stmt = conn.prepareStatement(INSERT_SQL) var pending = 0 rows.foreach { row => stmt.setDate(1, row.getAs[java.sql.Date]("stat_date")) stmt.setString(2, row.getAs[String]("city")) stmt.setLong(3, row.getAs[Long]("order_id")) stmt.setBigDecimal(4, row.getAs[java.math.BigDecimal]("amount")) stmt.setTimestamp(5, row.getAs[java.sql.Timestamp]("update_time")) stmt.addBatch() pending += 1 if (pending >= BATCH_SIZE) { stmt.executeBatch() stmt.clearBatch() pending = 0 } } if (pending > 0) { stmt.executeBatch() stmt.clearBatch() } } finally { if (stmt != null) stmt.close() if (conn != null) conn.close() } }这段代码有几个设计意图值得说。第一，连接建立在foreachPartition内部，也就是每个 executor 的每个 task 建一次，不是每行建一次，也不是在 driver 上建一次。第二，PreparedStatement只创建一次然后复用，避免反复解析 SQL。第三，攒够BATCH_SIZE行才提交一次，最终产生多少个 part 大致等于"分区数 × 每个分区的批次数"，这个数量是可以预先算出来的。
按上面的配置算一下：假设总共 3000 万行，8 个分区，每分区约 375 万行，每 5 万行提交一次，那每个分区产生 75 个批次，全表 600 个 part。这个数字超出了 MergeTree 默认的parts_to_throw_insert（300），所以要么把BATCH_SIZE提到 10 万以上，要么把分区数降到 4，要么在 CK 侧调大这个阈值。我一般的做法是反推：目标 part 数控制在 100 以内，那么分区数 × 每分区批次数 ≤ 100，在分区数固定为 8 的情况下，每分区最多 12 个批次，3000 万行除以 8 再除以 12，每批次应该在 30 万行以上。这就是为什么实际生产里批次往往开得比想象中大得多。
注意：
executeBatch之后一定要clearBatch，否则已经提交的语句会一直留在批处理缓存里，内存缓慢上涨，最后 OOM。这个坑很隐蔽，因为作业能跑完，只是 GC 压力越来越大。
4.3 幂等与重复数据：重跑场景必须提前设计
批处理任务一定会重跑，这不用怀疑。重跑的后果是同一批数据被写两次，如果表引擎是普通的MergeTree，查询结果就翻倍了。解决路径有三条，选哪条取决于业务对重复的容忍度。
第一条是用ReplacingMergeTree(update_time)配合排序键做最终去重。同一个排序键只保留update_time最大的那行。但要注意它是后台异步合并，查询的瞬间可能还没合并完，所以查询要写FINAL或者用argMax聚合。FINAL的代价是查询变慢，数据量大时很明显。
第二条是在写入前先删。ClickHouse 支持ALTER TABLE ... DELETE WHERE stat_date = 'xxx'，这是 mutation 操作，异步执行，返回不代表删完了。稳妥的做法是删完轮询system.mutations表确认完成再写：
SELECT count() FROM system.mutations WHERE table = 'dws_order_daily' AND is_done = 0;第三条是分区替换，也就是把当天数据写到一个临时表，然后用ALTER TABLE ... REPLACE PARTITION原子替换。这是最干净的做法，但是要求数据正好按分区组织，且临时表结构完全一致。
我一般按数据量选：小表用第一条省事；中大表用第三条最可靠；只有不方便按分区的场景才用第二条。
4.4 分布式表写入与本地表直写
如果 ClickHouse 是集群部署，写入目标有两种选择：写分布式表，或者按分片直接写各节点的本地表。
写分布式表的优点是简单，Spark 不用关心集群拓扑，CK 内部会按分片键分发。缺点是多了一跳转发，入口节点要接收全部数据再转发，网络和内存压力都集中在它身上；而且如果入口节点挂了，整个写入就失败了。
直写本地表的做法是让 Spark 感知分片拓扑，用repartition配合partitionId把数据分到对应节点。原生 JDBC 做这件事比较别扭，因为一个foreachPartition里只能用同一个连接串。可行的做法是按分片数分区，然后在每个分区里根据TaskContext.getPartitionId()查表选连接串：
val shardHosts = Array("ck-node-01", "ck-node-02", "ck-node-03", "ck-node-04") resultDf.repartition(shardHosts.length).foreachPartition { rows => val pid = org.apache.spark.TaskContext.getPartitionContext().partitionId() val host = shardHosts(pid % shardHosts.length) // 用 host 拼连接串，写本地表 }这里的数据分发是按分区号取模，跟 ClickHouse 的分片键不一定一致，所以严格来说只有单副本或者分片键就是分区键的场景才完全正确。要真正做到按分片键路由，得在写入前按分片键算哈希再分区，官方 Spark 连接器就是干这个事的。数据量大到需要精细控制分片的时候，直接上官方连接器更省心。
5. 性能调优与参数计算实录
调优这件事，最怕的是没测就凭感觉调。我习惯先用小数据量跑一遍，记录每个阶段的耗时，找出真正的瓶颈，再决定改哪里。写入链路的耗时通常分布在三块：Spark 侧的计算和 shuffle、序列化和网络传输、ClickHouse 侧的解析和落盘。用 Spark UI 的 stage 时间能看出前两块，CK 侧的耗时查system.query_log里的query_duration_ms和written_rows。
5.1 单批次大小的计算过程
批次大小不是拍脑袋定的，可以从两侧的约束反推。Spark 侧约束是单批次占用的堆内存，CK 侧约束是单次 INSERT 的内存和 part 大小。
先算 Spark 侧。假设一行数据序列化后约 200 字节，10 万行就是 20MB。PreparedStatement的批处理缓存会持有这些参数的引用，加上驱动内部的缓冲，实际占用大概是原始大小的 2 到 3 倍，也就是 40 到 60MB。如果 executor 的堆是 8GB，单个 task 的可用内存约 2GB，那 10 万行一批完全没压力，甚至 50 万行一批也扛得住。
再算 CK 侧。max_insert_block_size默认 1048576 行，min_insert_block_size_bytes默认 256MB，这两个参数决定了 CK 多久把内存里的数据刷成一个 part。如果每次 INSERT 只有 5 万行、10MB，那每次都会立刻形成一个 part，因为远低于阈值。这就是为什么小批次会产生大量 part 的根源。
综合两侧，我的经验值是这样的：
| 单次写入总行数 | 建议 Spark 分区数 | 建议批次大小 | 预计 part 数 |
|---|---|---|---|
| < 100 万 | 1 ~ 2 | 20 万 | < 10 |
| 100 万 ~ 1000 万 | 4 | 30 万 ~ 50 万 | 20 ~ 35 |
| 1000 万 ~ 1 亿 | 8 | 50 万 ~ 100 万 | 15 ~ 25 |
| > 1 亿 | 16 ~ 32 | 100 万 | 30 ~ 60 |
核心思路是让 part 数量保持在几十的量级，而不是几百上千。part 太多会让后台合并线程疲于奔命，查询时也要打开更多文件，性能全面下降。
5.2 Spark 侧资源参数与 CK 侧保护的配合
写入阶段的 Spark 资源不用给得太豪华，因为瓶颈通常不在 CPU。我一般这样配：
spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-memory 8g \ --executor-cores 4 \ --num-executors 8 \ --conf spark.sql.shuffle.partitions=64 \ --conf spark.speculation=false \ --conf spark.sql.adaptive.enabled=true \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --class com.example.CkWriteJob \ job.jarspark.speculation=false这一条必须加。推测执行会在某个 task 变慢时启动副本，副本也会写数据，结果就是重复写入或者连接数翻倍。批处理写入任务天生不适合开推测执行。
spark.sql.shuffle.partitions=64是给后面的repartition(8)留余地，如果直接设成 8，前面的计算阶段并行度太低会拖慢整体。中间的 repartition 会做一次全量 shuffle，64 到 8 的收敛过程代价不大。
CK 侧对应的保护参数：
<max_concurrent_queries>200</max_concurrent_queries> <max_insert_threads>4</max_insert_threads> <background_pool_size>16</background_pool_size> <parts_to_throw_insert>600</parts_to_throw_insert>max_insert_threads调到 4 能让单次 INSERT 的解析和落盘并行起来，但对小批次没意义，只有批次足够大才有收益。parts_to_throw_insert调大是治标不治本，我把它当成一个安全阀，正常链路不应该触发它。真正触发的时候，说明批次设计有问题，应该去改代码而不是继续调大阈值。
5.3 表结构对写入速度的影响
有一个容易被忽略的事实：表结构设计会直接影响写入速度。几个具体的点：
分区键的选择。PARTITION BY toYYYYMM(stat_date)意味着每个月一个分区，写入时如果数据跨越很多个月，每个分区都要单独写文件。如果按天分区PARTITION BY stat_date，分区数会多很多，小分区文件也多，反而更慢。一般按月分区是合适的，数据量特别大的表可以按周。
排序键的宽度。排序键的字段越多，写入时构建索引的成本越高。如果排序键有 5 个字段，每次写入都要维护这 5 个字段的稀疏索引，比 2 个字段慢不少。排序键按查询需求精简，不要什么字段都往里塞。
LowCardinality的使用。对于取值基数低于一万的字符串列（比如城市、渠道、状态），用LowCardinality(String)能显著减小存储体积和内存占用，写入时也有字典编码的开销，但总体是赚的。基数高的列（比如用户 ID）千万别用，字典会膨胀得比原始数据还大。
物化视图。如果表上挂了物化视图，每次写入都会触发视图的写入，写入耗时可能翻倍。这不是说不能用，而是要知道这个成本，写入密集的时候可以考虑先把视图 detach 掉，写完再 attach 并回填。
6. 常见问题排查速查表
这一节是我自己整理的排查手册，按现象分类。每次遇到新问题就往里加一条，时间长了发现大部分故障就那么十几种。
6.1 连接类异常
| 现象 | 可能原因 | 排查与解决 |
|---|---|---|
No suitable driver found | 驱动未打进 jar 或 scope 写成 provided | 检查 shade 插件配置，确认META-INF/services已合并 |
ClassNotFoundException: ru.yandex... | 驱动版本升级后包名变更 | 改用com.clickhouse.jdbc.ClickHouseDriver |
Connection refused | 端口不对或防火墙 | 8123 是 HTTP，9000 是 native，确认走哪个 |
TOO_MANY_SIMULTANEOUS_QUERIES | Spark 并发超过 CK 上限 | 降 repartition 数或调大max_concurrent_queries |
Authentication failed | 密码或用户权限 | 检查system.users，确认 INSERT 权限已授予 |
| 连接偶发超时 | 网络抖动或 CK 负载高 | 连接串加socket_timeout和connect_timeout，并加应用层重试 |
连接类问题里最值得说的是最后一个。批处理作业跑几个小时，中间偶尔一次网络抖动导致 task 失败，如果开了spark.task.maxFailures的重试，Spark 会自动重试成功的。但如果是 driver 端的连接超时，整个作业就挂了。所以socket_timeout不要设得太短，我一般设 600 秒，同时配合 3 次重试。
6.2 数据正确性问题
数据问题最讨厌，因为它不报错，只是结果不对，等你发现的时候可能已经过了好几天。
| 现象 | 可能原因 | 排查方法 |
|---|---|---|
| 时间戳差 8 小时 | 时区二次转换 | 检查连接串的use_time_zone，统一用 UTC |
| 大整数 ID 变负数 | UInt64超出LongType范围 | SQL 里转字符串，或确认 ID 不越界 |
| 金额精度丢失 | Decimal 被转成 Double | 全程用BigDecimal，不要在中间转 Double |
| 字符串末尾有空格 | FixedString(N)补零 | 读出来做trim，或建表改用String |
| 重跑后数据翻倍 | 缺少幂等设计 | 见 4.3 节，用替换分区或 ReplacingMergeTree |
| 中文乱码 | 字符集不一致 | 连接串加charset=UTF-8，建表确认是 UTF-8 |
提示：写入前做一次"空跑校验"能避免大部分数据问题。具体做法是在 Spark 里先
df.limit(100).collect()打印几行，人工看一眼字段对不对、时间对不对、金额有没有小数位错位。花三十秒，能省几小时。
6.3 写入阶段的典型报错
Too many parts是最高频的。前面讲过原理，这里说排查顺序：先看system.parts里这张表的 part 数，再看system.query_log里最近的 INSERT 语句频率，基本就能定位到是批次太小还是并发太高。
SELECT count() AS parts, sum(rows) AS total_rows FROM system.parts WHERE table = 'dws_order_daily' AND active = 1;Memory limit exceeded一般出现在单批次太大或者排序键太宽的时候。CK 的单个查询内存上限由max_memory_usage控制，调大能缓解，但根本办法还是控制批次大小。经验值是单次 INSERT 的数据量控制在 100MB 以内，超过就容易触发。
Cannot parse datetime这类报错通常出现在字符串转日期的边界值上，比如空字符串、0000-00-00这种非法日期。在 Spark 侧提前过滤掉这些脏数据，比在 CK 侧报错再回头找要高效得多。
Timeout exceeded在写入大表时出现，往往是 mutation 操作（ALTER DELETE）和 INSERT 抢资源。如果用了删除再写入的策略，务必确认删除的 mutation 已经is_done = 1再开始写。
7. 工程化落地的一些经验
代码能跑通只是第一步，真正上线还要处理调度、监控、告警、回滚这些事。这部分没有标准答案，我按自己的实践说一下。
7.1 调度与增量处理的设计
增量作业的核心是"水位线"。每张表维护一个处理到的位置，可以是日期、可以是自增 ID 的偏移量。我倾向于用日期，因为可读性好，出问题时容易人工干预。水位线存在哪？小规模用一张 CK 的元数据表就行，大规模用 Hive 表或者配置中心。
有个细节：水位线的更新必须在写入成功之后，而且要和写入在同一个事务语义里。CK 没有分布式事务，所以只能做补偿：写入成功后更新水位线，如果更新失败，下次重跑会重复写，这时候靠幂等机制兜住。这个顺序不能反，先更新水位线再写入，一旦写入失败就会丢数据。
重跑的设计要提前想好。我的做法是给作业加一个rerun_date参数，指定日期时只处理那一天的逻辑分区，并且写入前先清理当天的数据。这样运营或者数据同学发现某天数据有问题，改完代码直接重跑那一天，不影响其它分区。
7.2 监控指标与告警阈值
需要盯的指标不多，但每个都要真的设阈值：
| 指标 | 采集方式 | 告警阈值 |
|---|---|---|
| 作业耗时 | 调度系统记录 | 超过历史 P95 的 1.5 倍 |
| 写入行数 | 从 CK 的system.query_log查 | 与上游产出行数偏差超 1% |
| part 数量 | system.parts聚合 | 单表超过 300 |
| 查询失败率 | system.query_log里exception非空的占比 | 超过 0.5% |
| CK 磁盘使用率 | system.disks | 超过 80% |
"写入行数与上游偏差"这个指标特别有用，它能捕捉到一类静默失败：作业跑完了，状态是成功，但实际只写了一半数据。原因可能是某个 task 静默失败被 Spark 吞了，也可能是过滤条件写错。做一次行数对账，几秒钟的事，能挡住很多问题。
7.3 代码组织上的几点建议
第一，把连接配置抽成单独的配置类或配置文件，不要把地址、账号、密码散落在各个作业里。换一个集群地址要改十个文件，这种事发生过一次就够了。
第二，读和写封装成通用的工具方法，作业代码只负责业务逻辑。我的习惯是搞一个ClickHouseReader和ClickHouseWriter，参数包括表名、字段映射、批次大小、分区数，业务代码调用一行就完了。这样做的好处是攒批逻辑只有一份，调优只改一处。
第三，字段映射不要靠位置。INSERT INTO t VALUES (?, ?)这种写法，一旦表结构加了字段，位置就全乱了。老老实实写列名，虽然长一点，但改表的时候不会出事故。同理，select的时候也别用select *，字段清单明确写出来。
第四，给每个作业加一个 dry run 模式，只做计算不写入，把结果count一下打印出来。上线前跑一次，确认行数和字段都对，再真正写。这个开关在排查问题时也很有用，可以快速判断是计算错了还是写入错了。
最后分享一个我在实际使用中的体会：这条链路最脆弱的环节不是 Spark，也不是 ClickHouse，而是两者之间的网络和连接管理。计算逻辑写错了，跑一次就能看出来；连接管理出问题，往往是跑了几十分钟之后在某个 task 上偶发失败，排查起来非常折磨。所以在连接超时、重试、连接复用这几个地方多花点时间，回报率比优化计算逻辑高得多。我现在的习惯是，新接一个 ClickHouse 集群，先写一个最小化的读写连通性作业，把超时、重试、大字段、空值、特殊字符这些边界全跑一遍，确认没问题再往上堆业务逻辑。这个前置动作大概花半天，但能省掉后面好几天的救火时间。
